diff --git a/pkg/bus/types.go b/pkg/bus/types.go index e036b4ede..794db5b0f 100644 --- a/pkg/bus/types.go +++ b/pkg/bus/types.go @@ -57,6 +57,7 @@ type AudioChunk struct { SessionID string `json:"session_id"` SpeakerID string `json:"speaker_id"` // User ID or SSRC ChatID string `json:"chat_id"` // Where to respond + Channel string `json:"channel"` // Source channel type (e.g. "discord") Sequence uint64 `json:"sequence"` Timestamp uint32 `json:"timestamp"` SampleRate int `json:"sample_rate"` diff --git a/pkg/channels/discord/discord.go b/pkg/channels/discord/discord.go index c32a6409c..1db63c014 100644 --- a/pkg/channels/discord/discord.go +++ b/pkg/channels/discord/discord.go @@ -162,7 +162,7 @@ func (c *DiscordChannel) Send(ctx context.Context, msg bus.OutboundMessage) erro if c.cancelTTS != nil { c.cancelTTS() } - ttsCtx, ttsCancel := context.WithCancel(context.Background()) + ttsCtx, ttsCancel := context.WithCancel(c.ctx) c.cancelTTS = ttsCancel c.ttsMu.Unlock() @@ -651,7 +651,10 @@ func (c *DiscordChannel) listenVoiceControl(ctx context.Context) { select { case <-ctx.Done(): return - case ctrl := <-c.bus.VoiceControlsChan(): + case ctrl, ok := <-c.bus.VoiceControlsChan(): + if !ok { + return + } if ctrl.Type == "command" && ctrl.Action == "leave" { if strings.HasPrefix(ctrl.SessionID, "discord_vc_") { guildID := strings.TrimPrefix(ctrl.SessionID, "discord_vc_") diff --git a/pkg/channels/discord/voice.go b/pkg/channels/discord/voice.go index 8c4b84824..8dd1a6b55 100644 --- a/pkg/channels/discord/voice.go +++ b/pkg/channels/discord/voice.go @@ -1,7 +1,6 @@ package discord import ( - "bytes" "context" "fmt" "io" @@ -9,6 +8,7 @@ import ( "github.com/bwmarrin/discordgo" + "github.com/sipeed/picoclaw/pkg/audio" "github.com/sipeed/picoclaw/pkg/bus" "github.com/sipeed/picoclaw/pkg/logger" ) @@ -60,58 +60,14 @@ func streamOggOpusToDiscord(ctx context.Context, vc *discordgo.VoiceConnection, vc.Speaking(true) defer vc.Speaking(false) - var packet []byte - header := make([]byte, 27) - - for { - // Check for interruption + return audio.DecodeOggOpus(r, func(frame []byte) error { select { case <-ctx.Done(): return ctx.Err() - default: + case vc.OpusSend <- frame: + return nil } - - if _, err := io.ReadFull(r, header); err != nil { - if err == io.EOF || err == io.ErrUnexpectedEOF { - return nil - } - return fmt.Errorf("failed to read ogg header: %w", err) - } - if string(header[:4]) != "OggS" { - return fmt.Errorf("invalid ogg magic string") - } - - pageSegments := int(header[26]) - segmentTable := make([]byte, pageSegments) - if _, err := io.ReadFull(r, segmentTable); err != nil { - return fmt.Errorf("failed to read segment table: %w", err) - } - - for _, lacing := range segmentTable { - segment := make([]byte, lacing) - if _, err := io.ReadFull(r, segment); err != nil { - return fmt.Errorf("failed to read segment data: %w", err) - } - - packet = append(packet, segment...) - - // If lacing is less than 255, the packet is complete - if lacing < 255 { - if len(packet) > 0 { - // Ignore Ogg Opus headers - if !bytes.HasPrefix(packet, []byte("OpusHead")) && !bytes.HasPrefix(packet, []byte("OpusTags")) { - select { - case <-ctx.Done(): - return ctx.Err() - case vc.OpusSend <- packet: - } - } - // Start new packet - packet = nil - } - } - } - } + }) } func (c *DiscordChannel) receiveVoice(vc *discordgo.VoiceConnection, guildID string, chatID string) { @@ -193,6 +149,7 @@ func (c *DiscordChannel) receiveVoice(vc *discordgo.VoiceConnection, guildID str SessionID: sessionID, SpeakerID: fmt.Sprintf("%d", p.SSRC), ChatID: chatID, + Channel: "discord", Sequence: sequence, Timestamp: p.Timestamp, SampleRate: 48000, diff --git a/pkg/voice/agent.go b/pkg/voice/agent.go index 5ab19bc01..ceb14540e 100644 --- a/pkg/voice/agent.go +++ b/pkg/voice/agent.go @@ -26,6 +26,7 @@ type speechAccumulator struct { chatID string speakerID string sessionID string + channel string } func (a *speechAccumulator) Push(chunk bus.AudioChunk) { @@ -117,6 +118,7 @@ func (a *Agent) handleChunk(chunk bus.AudioChunk) { chatID: chunk.ChatID, speakerID: chunk.SpeakerID, sessionID: chunk.SessionID, + channel: chunk.Channel, } a.sessions[key] = acc logger.DebugCF("voice-agent", "Started accumulating voice", map[string]any{"key": key, "file": filename}) @@ -185,7 +187,10 @@ func (a *Agent) processUtterance(ctx context.Context, acc *speechAccumulator) { logger.InfoCF("voice-agent", "Transcription result", map[string]any{"text": res.Text, "duration": res.Duration}) - channelType := "discord" + channelType := acc.channel + if channelType == "" { + channelType = "discord" // fallback for legacy chunks + } text := strings.ToLower(strings.TrimSpace(res.Text)) if strings.Contains(text, "leave the voice channel") || strings.Contains(text, "leave voice") ||