From eb5114a65d820ced41afbcf39e89e14f63cdde29 Mon Sep 17 00:00:00 2001 From: Huaaudio Date: Mon, 23 Mar 2026 16:08:20 +0100 Subject: [PATCH] Update suggested fixes --- pkg/agent/loop.go | 3 ++ pkg/channels/discord/discord.go | 26 +++++------ pkg/channels/discord/voice.go | 80 +++++++++++++++++++++++++++++++-- pkg/gateway/gateway.go | 17 ++++--- pkg/tools/tts_send.go | 7 ++- pkg/tts/tts.go | 5 ++- pkg/voice/agent.go | 9 ++-- 7 files changed, 113 insertions(+), 34 deletions(-) diff --git a/pkg/agent/loop.go b/pkg/agent/loop.go index a3a0b328d..f120828c7 100644 --- a/pkg/agent/loop.go +++ b/pkg/agent/loop.go @@ -2443,6 +2443,9 @@ turnLoop: Channel: ts.channel, ChatID: ts.chatID, Content: toolResult.ForUser, + Metadata: map[string]string{ + "is_tool_call": "true", + }, }) logger.DebugCF("agent", "Sent tool result to user", map[string]any{ diff --git a/pkg/channels/discord/discord.go b/pkg/channels/discord/discord.go index 2dea2373e..a6ff8a332 100644 --- a/pkg/channels/discord/discord.go +++ b/pkg/channels/discord/discord.go @@ -47,10 +47,13 @@ type DiscordChannel struct { botUserID string // stored for mention checking bus *bus.MessageBus tts tts.TTSProvider + voiceMu sync.RWMutex + voiceSSRC map[string]map[uint32]string // guildID -> ssrc -> userID // TTS interruption: cancel active playback when user speaks ttsMu sync.Mutex cancelTTS context.CancelFunc + ttsPlayID uint64 } func NewDiscordChannel(cfg config.DiscordConfig, bus *bus.MessageBus) (*DiscordChannel, error) { @@ -83,7 +86,7 @@ func NewDiscordChannel(cfg config.DiscordConfig, bus *bus.MessageBus) (*DiscordC ctx: context.Background(), typingStop: make(map[string]chan struct{}), bus: bus, - bus: bus, + voiceSSRC: make(map[string]map[uint32]string), }, nil } @@ -155,14 +158,7 @@ func (c *DiscordChannel) Send(ctx context.Context, msg bus.OutboundMessage) erro return nil } - isToolCall := false - if msg.Metadata != nil { - if val, ok := msg.Metadata["is_tool_call"]; ok && val == "true" { - isToolCall = true - } - } - - if c.tts != nil && !isToolCall { + 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 { // Cancel any previous TTS playback @@ -171,10 +167,12 @@ func (c *DiscordChannel) Send(ctx context.Context, msg bus.OutboundMessage) erro c.cancelTTS() } ttsCtx, ttsCancel := context.WithCancel(c.ctx) + c.ttsPlayID++ + playID := c.ttsPlayID c.cancelTTS = ttsCancel c.ttsMu.Unlock() - go c.playTTS(ttsCtx, vc, msg.Content) + go c.playTTS(ttsCtx, vc, msg.Content, playID) } } } @@ -677,17 +675,13 @@ func (c *DiscordChannel) listenVoiceControl(ctx context.Context) { } } -func (c *DiscordChannel) playTTS(ctx context.Context, vc *discordgo.VoiceConnection, text string) { +func (c *DiscordChannel) playTTS(ctx context.Context, vc *discordgo.VoiceConnection, text string, playID uint64) { // Capture the cancel func associated with this playback (if any). - c.ttsMu.Lock() - playbackCancel := c.cancelTTS - c.ttsMu.Unlock() - // Clear cancelTTS when playback finishes (normal or interrupted), // but only if it still refers to this playback's cancel func. defer func() { c.ttsMu.Lock() - if c.cancelTTS == playbackCancel { + if c.ttsPlayID == playID { c.cancelTTS = nil } c.ttsMu.Unlock() diff --git a/pkg/channels/discord/voice.go b/pkg/channels/discord/voice.go index fdcc1adda..9d7597bd2 100644 --- a/pkg/channels/discord/voice.go +++ b/pkg/channels/discord/voice.go @@ -10,14 +10,45 @@ import ( "github.com/sipeed/picoclaw/pkg/audio" "github.com/sipeed/picoclaw/pkg/bus" + "github.com/sipeed/picoclaw/pkg/identity" "github.com/sipeed/picoclaw/pkg/logger" ) +func (c *DiscordChannel) setVoiceUserID(guildID string, ssrc uint32, userID string) { + if userID == "" { + return + } + + c.voiceMu.Lock() + defer c.voiceMu.Unlock() + + ssrcMap, ok := c.voiceSSRC[guildID] + if !ok { + ssrcMap = make(map[uint32]string) + c.voiceSSRC[guildID] = ssrcMap + } + ssrcMap[ssrc] = userID +} + +func (c *DiscordChannel) voiceUserID(guildID string, ssrc uint32) string { + c.voiceMu.RLock() + defer c.voiceMu.RUnlock() + + ssrcMap, ok := c.voiceSSRC[guildID] + if !ok { + return "" + } + return ssrcMap[ssrc] +} + func (c *DiscordChannel) handleVoiceCommand(s *discordgo.Session, m *discordgo.MessageCreate) bool { if m.Content == "!vc join" { vs, err := s.State.VoiceState(m.GuildID, m.Author.ID) if err != nil || vs == nil { - if _, sendErr := s.ChannelMessageSend(m.ChannelID, "You need to be in a voice channel first!"); sendErr != nil { + if _, sendErr := s.ChannelMessageSend( + m.ChannelID, + "You need to be in a voice channel first!", + ); sendErr != nil { logger.InfoCF("discord", "Failed to send voice channel requirement message", map[string]any{ "channel": m.ChannelID, "error": sendErr, @@ -30,7 +61,10 @@ func (c *DiscordChannel) handleVoiceCommand(s *discordgo.Session, m *discordgo.M vc, err := s.ChannelVoiceJoin(c.ctx, m.GuildID, vs.ChannelID, false, false) vc, err := s.ChannelVoiceJoin(c.ctx, m.GuildID, vs.ChannelID, false, false) if err != nil { - if _, sendErr := s.ChannelMessageSend(m.ChannelID, fmt.Sprintf("Failed to join voice channel: %v", err)); sendErr != nil { + if _, sendErr := s.ChannelMessageSend( + m.ChannelID, + fmt.Sprintf("Failed to join voice channel: %v", err), + ); sendErr != nil { logger.InfoCF("discord", "Failed to send voice join error message", map[string]any{ "channel": m.ChannelID, "error": sendErr, @@ -40,7 +74,10 @@ func (c *DiscordChannel) handleVoiceCommand(s *discordgo.Session, m *discordgo.M } go c.receiveVoice(vc, m.GuildID, m.ChannelID) - if _, sendErr := s.ChannelMessageSend(m.ChannelID, "Joined Voice Channel! Listening for audio..."); sendErr != nil { + if _, sendErr := s.ChannelMessageSend( + m.ChannelID, + "Joined Voice Channel! Listening for audio...", + ); sendErr != nil { logger.InfoCF("discord", "Failed to send voice join success message", map[string]any{ "channel": m.ChannelID, "error": sendErr, @@ -104,6 +141,19 @@ func streamOggOpusToDiscord(ctx context.Context, vc *discordgo.VoiceConnection, func (c *DiscordChannel) receiveVoice(vc *discordgo.VoiceConnection, guildID string, chatID string) { logger.InfoCF("discord", "Started listening for voice", map[string]any{"guild": guildID}) + vc.AddHandler(func(_ *discordgo.VoiceConnection, vs *discordgo.VoiceSpeakingUpdate) { + if vs == nil { + return + } + c.setVoiceUserID(guildID, uint32(vs.SSRC), vs.UserID) + }) + + defer func() { + c.voiceMu.Lock() + delete(c.voiceSSRC, guildID) + c.voiceMu.Unlock() + }() + go func(ctx context.Context, vc *discordgo.VoiceConnection) { // Recover from potential panics if OpusSend is closed mid-send. defer func() { @@ -211,11 +261,33 @@ func (c *DiscordChannel) receiveVoice(vc *discordgo.VoiceConnection, guildID str interruptCount = 0 } + userID := c.voiceUserID(guildID, p.SSRC) + if userID == "" { + logger.DebugCF("discord", "Dropping voice packet without user mapping", map[string]any{ + "ssrc": p.SSRC, + "guild": guildID, + }) + continue + } + + sender := bus.SenderInfo{ + Platform: "discord", + PlatformID: userID, + CanonicalID: identity.BuildCanonicalID("discord", userID), + } + if !c.IsAllowedSender(sender) { + logger.DebugCF("discord", "Voice packet rejected by allowlist", map[string]any{ + "user_id": userID, + "guild": guildID, + }) + continue + } + sequence++ chunk := bus.AudioChunk{ SessionID: sessionID, - SpeakerID: fmt.Sprintf("%d", p.SSRC), + SpeakerID: userID, ChatID: chatID, Channel: "discord", Sequence: sequence, diff --git a/pkg/gateway/gateway.go b/pkg/gateway/gateway.go index ad7591e3a..d2f821f8f 100644 --- a/pkg/gateway/gateway.go +++ b/pkg/gateway/gateway.go @@ -304,15 +304,10 @@ func setupAndStartServices( agentLoop.SetChannelManager(runningServices.ChannelManager) agentLoop.SetMediaStore(runningServices.MediaStore) - if transcriber := asr.DetectTranscriber(cfg); transcriber != nil { + transcriber := asr.DetectTranscriber(cfg) + if transcriber != nil { agentLoop.SetTranscriber(transcriber) logger.InfoCF("voice", "Transcription enabled (agent-level)", map[string]any{"provider": transcriber.Name()}) - - // Start Voice Agent Orchestrator - vaCtx, vaCancel := context.WithCancel(context.Background()) - runningServices.VoiceAgentCancel = vaCancel - voiceAgent := voice.NewAgent(msgBus, transcriber) - voiceAgent.Start(vaCtx) } enabledChannels := runningServices.ChannelManager.GetEnabledChannels() @@ -330,6 +325,14 @@ func setupAndStartServices( return nil, fmt.Errorf("error starting channels: %w", err) } + if transcriber != nil { + // Start Voice Agent Orchestrator after channels are ready. + vaCtx, vaCancel := context.WithCancel(context.Background()) + runningServices.VoiceAgentCancel = vaCancel + voiceAgent := voice.NewAgent(msgBus, transcriber) + voiceAgent.Start(vaCtx) + } + fmt.Printf( "✓ Health endpoints available at http://%s:%d/health, /ready and /reload (POST)\n", cfg.Gateway.Host, diff --git a/pkg/tools/tts_send.go b/pkg/tools/tts_send.go index 71dd7e593..27219630e 100644 --- a/pkg/tools/tts_send.go +++ b/pkg/tools/tts_send.go @@ -87,13 +87,18 @@ func (t *SendTTSTool) Execute(ctx context.Context, args map[string]any) *ToolRes if err != nil { return ErrorResult(fmt.Sprintf("failed to create temp file: %v", err)).WithError(err) } - defer file.Close() _, err = io.Copy(file, stream) if err != nil { + file.Close() return ErrorResult(fmt.Sprintf("failed to write tts audio: %v", err)).WithError(err) } + err = file.Close() + if err != nil { + return ErrorResult(fmt.Sprintf("failed to close tts audio file: %v", err)).WithError(err) + } + filename, _ := args["filename"].(string) filename = strings.TrimSpace(filename) if filename == "" { diff --git a/pkg/tts/tts.go b/pkg/tts/tts.go index 07da18bb8..eca4d28ee 100644 --- a/pkg/tts/tts.go +++ b/pkg/tts/tts.go @@ -82,7 +82,10 @@ func NewOpenAITTSProvider(apiKey string, apiBase string, proxyURL string) *OpenA Proxy: http.ProxyURL(pURL), } } else { - logger.Warnf("NewOpenAITTSProvider: invalid proxy URL %q: %v; proceeding without proxy", proxyURL, err) + logger.WarnF( + "NewOpenAITTSProvider: invalid proxy URL; proceeding without proxy", + map[string]any{"proxyURL": proxyURL, "error": err}, + ) } } diff --git a/pkg/voice/agent.go b/pkg/voice/agent.go index a9b8cb9ec..2bcd13515 100644 --- a/pkg/voice/agent.go +++ b/pkg/voice/agent.go @@ -120,16 +120,15 @@ func (a *Agent) handleChunk(chunk bus.AudioChunk) { return } - a.mu.Lock() - defer a.mu.Unlock() - key := fmt.Sprintf("%s_%s", chunk.SessionID, chunk.SpeakerID) + a.mu.Lock() acc, exists := a.sessions[key] if !exists { filename := filepath.Join(os.TempDir(), fmt.Sprintf("voice_%s_%d.ogg", key, time.Now().UnixNano())) writer, err := oggwriter.New(filename, uint32(chunk.SampleRate), uint16(chunk.Channels)) if err != nil { + a.mu.Unlock() logger.ErrorCF("voice-agent", "Failed to create OggWriter", map[string]any{"error": err}) return } @@ -146,6 +145,7 @@ func (a *Agent) handleChunk(chunk bus.AudioChunk) { a.sessions[key] = acc logger.DebugCF("voice-agent", "Started accumulating voice", map[string]any{"key": key, "file": filename}) } + a.mu.Unlock() acc.Push(chunk) } @@ -246,8 +246,7 @@ func (a *Agent) processUtterance(ctx context.Context, acc *speechAccumulator) { Content: res.Text + oralPrompt, Peer: bus.Peer{Kind: "channel", ID: acc.chatID}, Metadata: map[string]string{ - "is_voice": "true", - "oral_prompt": oralPrompt, + "is_voice": "true", }, }); err != nil { logger.ErrorCF("voice-agent", "Failed to publish inbound message", map[string]any{"error": err})