diff --git a/.golangci.yaml b/.golangci.yaml index ea3107ec8..b2b772406 100644 --- a/.golangci.yaml +++ b/.golangci.yaml @@ -61,6 +61,9 @@ linters: - usestdlibvars - usetesting settings: + gomoddirectives: + replace-allow-list: + - github.com/bwmarrin/discordgo errcheck: check-type-assertions: true check-blank: true diff --git a/config/config.example.json b/config/config.example.json index ff2969dcb..f41614452 100644 --- a/config/config.example.json +++ b/config/config.example.json @@ -414,6 +414,9 @@ "read_file": { "enabled": true }, + "send_tts": { + "enabled": false + }, "spawn": { "enabled": true }, diff --git a/go.mod b/go.mod index 54c275102..39d33bbd5 100644 --- a/go.mod +++ b/go.mod @@ -26,6 +26,8 @@ require ( github.com/mymmrac/telego v1.7.0 github.com/open-dingtalk/dingtalk-stream-sdk-go v0.9.1 github.com/openai/openai-go/v3 v3.22.0 + github.com/pion/rtp v1.8.7 + github.com/pion/webrtc/v3 v3.3.6 github.com/rivo/tview v0.42.0 github.com/rs/zerolog v1.34.0 github.com/slack-go/slack v0.17.3 @@ -60,6 +62,7 @@ require ( github.com/aws/aws-sdk-go-v2/service/sts v1.41.9 // indirect github.com/aws/smithy-go v1.24.2 // indirect github.com/beeper/argo-go v1.1.2 // indirect + github.com/cloudflare/circl v1.6.3 // indirect github.com/coder/websocket v1.8.14 // indirect github.com/davecgh/go-spew v1.1.1 // indirect github.com/dustin/go-humanize v1.0.1 // indirect @@ -73,6 +76,7 @@ require ( github.com/mattn/go-sqlite3 v1.14.34 // indirect github.com/ncruces/go-strftime v1.0.0 // indirect github.com/petermattis/goid v0.0.0-20260226131333-17d1149c6ac6 // indirect + github.com/pion/randutil v0.1.0 // indirect github.com/pmezard/go-difflib v1.0.0 // indirect github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect github.com/rivo/uniseg v0.4.7 // indirect @@ -116,3 +120,5 @@ require ( golang.org/x/sync v0.20.0 // indirect golang.org/x/sys v0.42.0 ) + +replace github.com/bwmarrin/discordgo => github.com/yeongaori/discordgo-fork v0.0.0-20260319072544-e8e546f5d532 diff --git a/go.sum b/go.sum index ae12473f3..c6da69a71 100644 --- a/go.sum +++ b/go.sum @@ -51,8 +51,6 @@ github.com/aws/smithy-go v1.24.2 h1:FzA3bu/nt/vDvmnkg+R8Xl46gmzEDam6mZ1hzmwXFng= github.com/aws/smithy-go v1.24.2/go.mod h1:YE2RhdIuDbA5E5bTdciG9KrW3+TiEONeUWCqxX9i1Fc= github.com/beeper/argo-go v1.1.2 h1:UQI2G8F+NLfGTOmTUI0254pGKx/HUU/etbUGTJv91Fs= github.com/beeper/argo-go v1.1.2/go.mod h1:M+LJAnyowKVQ6Rdj6XYGEn+qcVFkb3R/MUpqkGR0hM4= -github.com/bwmarrin/discordgo v0.29.0 h1:FmWeXFaKUwrcL3Cx65c20bTRW+vOb6k8AnaP+EgjDno= -github.com/bwmarrin/discordgo v0.29.0/go.mod h1:NJZpH+1AfhIcyQsPeuBKsUtYrRnjkyu0kIVMCHkZtRY= github.com/bytedance/gopkg v0.1.3 h1:TPBSwH8RsouGCBcMBktLt1AymVo2TVsBVCY4b6TnZ/M= github.com/bytedance/gopkg v0.1.3/go.mod h1:576VvJ+eJgyCzdjS+c4+77QF3p7ubbtiKARP3TxducM= github.com/bytedance/sonic v1.15.0 h1:/PXeWFaR5ElNcVE84U0dOHjiMHQOwNIx3K4ymzh/uSE= @@ -63,6 +61,8 @@ github.com/caarlos0/env/v11 v11.4.0 h1:Kcb6t5kIIr4XkoQC9AF2j+8E1Jsrl3Wz/hhm1LtoG github.com/caarlos0/env/v11 v11.4.0/go.mod h1:qupehSf/Y0TUTsxKywqRt/vJjN5nz6vauiYEUUr8P4U= github.com/cespare/xxhash/v2 v2.1.2/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= github.com/cespare/xxhash/v2 v2.2.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= +github.com/cloudflare/circl v1.6.3 h1:9GPOhQGF9MCYUeXyMYlqTR6a5gTrgR/fBLXvUgtVcg8= +github.com/cloudflare/circl v1.6.3/go.mod h1:2eXP6Qfat4O/Yhh8BznvKnJ+uzEoTQ6jVKJRn81BiS4= github.com/cloudwego/base64x v0.1.6 h1:t11wG9AECkCDk5fMSoxmufanudBtJ+/HemLstXDLI2M= github.com/cloudwego/base64x v0.1.6/go.mod h1:OFcloc187FXDaYHvrNIjxSe8ncn0OOM8gEHfghB2IPU= github.com/coder/websocket v1.8.14 h1:9L0p0iKiNOibykf283eHkKUHHrpG7f65OE3BhhO7v9g= @@ -196,6 +196,12 @@ github.com/openai/openai-go/v3 v3.22.0 h1:6MEoNoV8sbjOVmXdvhmuX3BjVbVdcExbVyGixi github.com/openai/openai-go/v3 v3.22.0/go.mod h1:cdufnVK14cWcT9qA1rRtrXx4FTRsgbDPW7Ia7SS5cZo= github.com/petermattis/goid v0.0.0-20260226131333-17d1149c6ac6 h1:rh2lKw/P/EqHa724vYH2+VVQ1YnW4u6EOXl0PMAovZE= github.com/petermattis/goid v0.0.0-20260226131333-17d1149c6ac6/go.mod h1:pxMtw7cyUw6B2bRH0ZBANSPg+AoSud1I1iyJHI69jH4= +github.com/pion/randutil v0.1.0 h1:CFG1UdESneORglEsnimhUjf33Rwjubwj6xfiOXBa3mA= +github.com/pion/randutil v0.1.0/go.mod h1:XcJrSMMbbMRhASFVOlj/5hQial/Y8oH/HVo7TBZq+j8= +github.com/pion/rtp v1.8.7 h1:qslKkG8qxvQ7hqaxkmL7Pl0XcUm+/Er7nMnu6Vq+ZxM= +github.com/pion/rtp v1.8.7/go.mod h1:pBGHaFt/yW7bf1jjWAoUjpSNoDnw98KTMg+jWWvziqU= +github.com/pion/webrtc/v3 v3.3.6 h1:7XAh4RPtlY1Vul6/GmZrv7z+NnxKA6If0KStXBI2ZLE= +github.com/pion/webrtc/v3 v3.3.6/go.mod h1:zyN7th4mZpV27eXybfR/cnUf3J2DRy8zw/mdjD9JTNM= github.com/pkg/diff v0.0.0-20210226163009-20ebb0f2a09e/go.mod h1:pJLUxLENpZxwdsKMEsNbx1VGcRFpLqf3715MtcvvzbA= github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= @@ -264,6 +270,8 @@ github.com/vektah/gqlparser/v2 v2.5.27 h1:RHPD3JOplpk5mP5JGX8RKZkt2/Vwj/PZv0HxTd github.com/vektah/gqlparser/v2 v2.5.27/go.mod h1:D1/VCZtV3LPnQrcPBeR/q5jkSQIPti0uYCP/RI0gIeo= github.com/xyproto/randomstring v1.0.5 h1:YtlWPoRdgMu3NZtP45drfy1GKoojuR7hmRcnhZqKjWU= github.com/xyproto/randomstring v1.0.5/go.mod h1:rgmS5DeNXLivK7YprL0pY+lTuhNQW3iGxZ18UQApw/E= +github.com/yeongaori/discordgo-fork v0.0.0-20260319072544-e8e546f5d532 h1:gxFHYeUDGziRb0zXYEqBFohC+NJbIW9L0tddaXMWr2o= +github.com/yeongaori/discordgo-fork v0.0.0-20260319072544-e8e546f5d532/go.mod h1:A0FcMFJKJ9fRjgSuZ2o+pIQ6mPS81SVuiLN2vYTa7Ao= github.com/yosida95/uritemplate/v3 v3.0.2 h1:Ed3Oyj9yrmi9087+NczuL5BwkIc4wvTb5zIM+UJPGz4= github.com/yosida95/uritemplate/v3 v3.0.2/go.mod h1:ILOh0sOhIJR3+L/8afwt/kE++YT040gmv5BQTMR2HP4= github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= @@ -283,7 +291,6 @@ golang.org/x/arch v0.24.0/go.mod h1:dNHoOeKiyja7GTvF9NJS1l3Z2yntpQNzgrjh1cU103A= golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI= golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto= -golang.org/x/crypto v0.0.0-20210421170649-83a5a9bb288b/go.mod h1:T9bdIzuCu7OtxOm1hfPfRQxPLYneinmdGuTeoZ9dtd4= golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc= golang.org/x/crypto v0.16.0/go.mod h1:gCAAfMLgwOJRpTjQ2zCCt2OcSfYMTeZVSRtQlPC7Nq4= golang.org/x/crypto v0.49.0 h1:+Ng2ULVvLHnJ/ZFEq4KdcDd/cfjrrjjNSXNzxg0Y4U4= diff --git a/pkg/agent/loop.go b/pkg/agent/loop.go index fb9edda25..3c592092d 100644 --- a/pkg/agent/loop.go +++ b/pkg/agent/loop.go @@ -18,6 +18,8 @@ import ( "sync/atomic" "time" + "github.com/sipeed/picoclaw/pkg/audio/asr" + "github.com/sipeed/picoclaw/pkg/audio/tts" "github.com/sipeed/picoclaw/pkg/bus" "github.com/sipeed/picoclaw/pkg/channels" "github.com/sipeed/picoclaw/pkg/commands" @@ -31,7 +33,6 @@ import ( "github.com/sipeed/picoclaw/pkg/state" "github.com/sipeed/picoclaw/pkg/tools" "github.com/sipeed/picoclaw/pkg/utils" - "github.com/sipeed/picoclaw/pkg/voice" ) type AgentLoop struct { @@ -51,7 +52,7 @@ type AgentLoop struct { fallback *providers.FallbackChain channelManager *channels.Manager mediaStore media.MediaStore - transcriber voice.Transcriber + transcriber asr.Transcriber cmdRegistry *commands.Registry mcp mcpRuntime hookRuntime hookRuntime @@ -156,6 +157,13 @@ func registerSharedTools( provider providers.LLMProvider, ) { allowReadPaths := buildAllowReadPatterns(cfg) + var ttsProvider tts.TTSProvider + if cfg.Tools.IsToolEnabled("send_tts") { + ttsProvider = tts.DetectTTS(cfg) + if ttsProvider == nil { + logger.WarnCF("voice-tts", "send_tts enabled but no TTS provider configured", nil) + } + } for _, agentID := range registry.ListAgentIDs() { agent, ok := registry.GetAgent(agentID) @@ -246,6 +254,10 @@ func registerSharedTools( agent.Tools.Register(sendFileTool) } + if ttsProvider != nil { + agent.Tools.Register(tools.NewSendTTSTool(ttsProvider, nil)) + } + // Skill discovery and installation tools skills_enabled := cfg.Tools.IsToolEnabled("skills") find_skills_enable := cfg.Tools.IsToolEnabled("find_skills") @@ -1033,10 +1045,15 @@ func (al *AgentLoop) SetMediaStore(s media.MediaStore) { agent.Tools.SetMediaStore(s) } } + registry.ForEachTool("send_tts", func(t tools.Tool) { + if st, ok := t.(*tools.SendTTSTool); ok { + st.SetMediaStore(s) + } + }) } // SetTranscriber injects a voice transcriber for agent-level audio transcription. -func (al *AgentLoop) SetTranscriber(t voice.Transcriber) { +func (al *AgentLoop) SetTranscriber(t asr.Transcriber) { al.transcriber = t } @@ -1057,19 +1074,23 @@ func (al *AgentLoop) transcribeAudioInMessage(ctx context.Context, msg bus.Inbou // Transcribe each audio media ref in order. var transcriptions []string + var keptMedia []string for _, ref := range msg.Media { path, meta, err := al.mediaStore.ResolveWithMeta(ref) if err != nil { logger.WarnCF("voice", "Failed to resolve media ref", map[string]any{"ref": ref, "error": err}) + keptMedia = append(keptMedia, ref) continue } if !utils.IsAudioFile(meta.Filename, meta.ContentType) { + keptMedia = append(keptMedia, ref) continue } result, err := al.transcriber.Transcribe(ctx, path) if err != nil { logger.WarnCF("voice", "Transcription failed", map[string]any{"ref": ref, "error": err}) transcriptions = append(transcriptions, "") + keptMedia = append(keptMedia, ref) continue } transcriptions = append(transcriptions, result.Text) @@ -1089,15 +1110,21 @@ func (al *AgentLoop) transcribeAudioInMessage(ctx context.Context, msg bus.Inbou } text := transcriptions[idx] idx++ + if text == "" { + return match + } return "[voice: " + text + "]" }) // Append any remaining transcriptions not matched by an annotation. for ; idx < len(transcriptions); idx++ { - newContent += "\n[voice: " + transcriptions[idx] + "]" + if transcriptions[idx] != "" { + newContent += "\n[voice: " + transcriptions[idx] + "]" + } } msg.Content = newContent + msg.Media = keptMedia return msg, true } @@ -2425,6 +2452,28 @@ turnLoop: if toolResult == nil { toolResult = tools.ErrorResult("hook returned nil tool result") } + + // Send ForUser if not silent and has content. + // For ResponseHandled tools, send regardless of SendResponse setting, + // since they've already handled the response (e.g., send_tts, send_file). + shouldSendForUser := !toolResult.Silent && toolResult.ForUser != "" && + (ts.opts.SendResponse || toolResult.ResponseHandled) + if shouldSendForUser { + al.bus.PublishOutbound(ctx, bus.OutboundMessage{ + 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{ + "tool": toolName, + "content_len": len(toolResult.ForUser), + }) + } + if len(toolResult.Media) > 0 && toolResult.ResponseHandled { parts := make([]bus.MediaPart, 0, len(toolResult.Media)) for _, ref := range toolResult.Media { @@ -2470,19 +2519,6 @@ turnLoop: allResponsesHandled = false } - if !toolResult.Silent && toolResult.ForUser != "" && ts.opts.SendResponse { - al.bus.PublishOutbound(ctx, bus.OutboundMessage{ - Channel: ts.channel, - ChatID: ts.chatID, - Content: toolResult.ForUser, - }) - logger.DebugCF("agent", "Sent tool result to user", - map[string]any{ - "tool": toolName, - "content_len": len(toolResult.ForUser), - }) - } - contentForLLM := toolResult.ContentForLLM() // Filter sensitive data (API keys, tokens, secrets) before sending to LLM diff --git a/pkg/audio/asr/agent.go b/pkg/audio/asr/agent.go new file mode 100644 index 000000000..32ce0c92a --- /dev/null +++ b/pkg/audio/asr/agent.go @@ -0,0 +1,252 @@ +package asr + +import ( + "context" + "fmt" + "os" + "path/filepath" + "strings" + "sync" + "time" + + "github.com/pion/rtp" + "github.com/pion/webrtc/v3/pkg/media/oggwriter" + + "github.com/sipeed/picoclaw/pkg/bus" + "github.com/sipeed/picoclaw/pkg/logger" +) + +type speechAccumulator struct { + writer *oggwriter.OggWriter + file string + lastAudioAt time.Time + mu sync.Mutex + closed bool + chatID string + speakerID string + sessionID string + channel string +} + +func (a *speechAccumulator) Push(chunk bus.AudioChunk) { + a.mu.Lock() + defer a.mu.Unlock() + + if a.closed { + return + } + + a.lastAudioAt = time.Now() + + pkt := &rtp.Packet{ + Header: rtp.Header{ + SequenceNumber: uint16(chunk.Sequence), + Timestamp: chunk.Timestamp, + SSRC: 1, // Stable arbitrary dummy + }, + Payload: chunk.Data, + } + + if err := a.writer.WriteRTP(pkt); err != nil { + logger.ErrorCF("voice-agent", "Failed to write RTP", map[string]any{"error": err}) + } +} + +func (a *speechAccumulator) Close() { + a.mu.Lock() + defer a.mu.Unlock() + if !a.closed { + a.writer.Close() + a.closed = true + } +} + +type Agent struct { + bus *bus.MessageBus + transcriber Transcriber + + mu sync.Mutex + sessions map[string]*speechAccumulator // keyed by sessionID_speakerID +} + +func NewAgent(mb *bus.MessageBus, t Transcriber) *Agent { + return &Agent{ + bus: mb, + transcriber: t, + sessions: make(map[string]*speechAccumulator), + } +} + +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) { + chunks := a.bus.AudioChunksChan() + for { + select { + case <-ctx.Done(): + return + case chunk, ok := <-chunks: + if !ok { + return + } + a.handleChunk(chunk) + } + } +} + +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 + } + + 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 + } + + acc = &speechAccumulator{ + writer: writer, + file: filename, + lastAudioAt: time.Now(), + 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}) + } + a.mu.Unlock() + + acc.Push(chunk) +} + +func (a *Agent) vadTick(ctx context.Context) { + ticker := time.NewTicker(500 * time.Millisecond) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + a.checkSilence(ctx) + } + } +} + +func (a *Agent) checkSilence(ctx context.Context) { + a.mu.Lock() + now := time.Now() + var finished []*speechAccumulator + + for key, acc := range a.sessions { + acc.mu.Lock() + last := acc.lastAudioAt + acc.mu.Unlock() + + if now.Sub(last) > 1500*time.Millisecond { + acc.Close() + delete(a.sessions, key) + finished = append(finished, acc) + } + } + a.mu.Unlock() + + for _, acc := range finished { + go a.processUtterance(ctx, acc) + } +} + +func (a *Agent) processUtterance(ctx context.Context, acc *speechAccumulator) { + defer os.Remove(acc.file) + + logger.InfoCF("voice-agent", "User finished speaking, transcribing...", map[string]any{"file": acc.file}) + + if a.transcriber == nil { + logger.ErrorCF("voice-agent", "No STT configured!", nil) + return + } + + res, err := a.transcriber.Transcribe(ctx, acc.file) + if err != nil { + logger.ErrorCF("voice-agent", "Transcription failed", map[string]any{"error": err}) + return + } + + if res.Text == "" { + logger.DebugCF("voice-agent", "Ignored empty transcription", map[string]any{"file": acc.file}) + return + } + + logger.InfoCF("voice-agent", "Transcription result", map[string]any{"text": res.Text, "duration": res.Duration}) + + 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") || + 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) + if err := a.bus.PublishVoiceControl(ctx, bus.VoiceControl{ + SessionID: acc.sessionID, + Type: "command", + Action: "leave", + }); 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." + + if err := a.bus.PublishInbound(ctx, bus.InboundMessage{ + Channel: channelType, + SenderID: acc.speakerID, + ChatID: acc.chatID, + Content: res.Text + oralPrompt, + Peer: bus.Peer{Kind: "channel", ID: acc.chatID}, + Metadata: map[string]string{ + "is_voice": "true", + }, + }); err != nil { + logger.ErrorCF("voice-agent", "Failed to publish inbound message", map[string]any{"error": err}) + } +} diff --git a/pkg/audio/asr/agent_test.go b/pkg/audio/asr/agent_test.go new file mode 100644 index 000000000..cc1b008a4 --- /dev/null +++ b/pkg/audio/asr/agent_test.go @@ -0,0 +1,196 @@ +package asr + +import ( + "context" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/pion/webrtc/v3/pkg/media/oggwriter" + + "github.com/sipeed/picoclaw/pkg/bus" +) + +type fakeTranscriber struct { + text string + err error + lastPath string +} + +func (f *fakeTranscriber) Name() string { return "fake" } + +func (f *fakeTranscriber) Transcribe(ctx context.Context, audioFilePath string) (*TranscriptionResponse, error) { + f.lastPath = audioFilePath + if f.err != nil { + return nil, f.err + } + return &TranscriptionResponse{Text: f.text}, nil +} + +func waitForFileRemoval(t *testing.T, path string, timeout time.Duration) { + t.Helper() + + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + if _, err := os.Stat(path); os.IsNotExist(err) { + return + } + time.Sleep(10 * time.Millisecond) + } + if _, err := os.Stat(path); err == nil { + t.Fatalf("expected file to be removed: %s", path) + } +} + +func TestAgentHandleChunkCreatesSession(t *testing.T) { + t.Parallel() + + mb := bus.NewMessageBus() + defer mb.Close() + + agent := NewAgent(mb, &fakeTranscriber{}) + + chunk := bus.AudioChunk{ + SessionID: "sess", + SpeakerID: "speaker", + ChatID: "chat", + Channel: "discord", + Sequence: 1, + Timestamp: 1, + SampleRate: 48000, + Channels: 2, + Format: "opus", + Data: []byte{0xF8, 0xFF, 0xFE}, + } + + agent.handleChunk(chunk) + + key := "sess_speaker" + agent.mu.Lock() + acc, ok := agent.sessions[key] + agent.mu.Unlock() + if !ok { + t.Fatal("expected session to be created") + } + + acc.Close() + _ = os.Remove(acc.file) +} + +func TestAgentHandleChunkIgnoresUnsupportedFormat(t *testing.T) { + t.Parallel() + + mb := bus.NewMessageBus() + defer mb.Close() + + agent := NewAgent(mb, &fakeTranscriber{}) + + chunk := bus.AudioChunk{Format: "pcm"} + agent.handleChunk(chunk) + + agent.mu.Lock() + count := len(agent.sessions) + agent.mu.Unlock() + if count != 0 { + t.Fatalf("expected no sessions, got %d", count) + } +} + +func TestAgentProcessUtteranceLeaveCommand(t *testing.T) { + t.Parallel() + + mb := bus.NewMessageBus() + defer mb.Close() + + tr := &fakeTranscriber{text: "please leave the voice channel now"} + agent := NewAgent(mb, tr) + + tmpDir := t.TempDir() + filePath := filepath.Join(tmpDir, "voice.ogg") + if err := os.WriteFile(filePath, []byte("data"), 0o600); err != nil { + t.Fatalf("write temp file: %v", err) + } + + acc := &speechAccumulator{ + file: filePath, + chatID: "chat", + speakerID: "speaker", + sessionID: "sess", + channel: "discord", + } + + agent.processUtterance(context.Background(), acc) + + select { + case ctrl := <-mb.VoiceControlsChan(): + if ctrl.Action != "leave" || ctrl.Type != "command" || ctrl.SessionID != "sess" { + t.Fatalf("unexpected voice control: %#v", ctrl) + } + case <-time.After(250 * time.Millisecond): + t.Fatal("expected voice control publish") + } + + select { + case out := <-mb.OutboundChan(): + if !strings.Contains(out.Content, "Leaving the voice channel") { + t.Fatalf("unexpected outbound content: %q", out.Content) + } + case <-time.After(250 * time.Millisecond): + t.Fatal("expected outbound publish") + } + + if _, err := os.Stat(filePath); !os.IsNotExist(err) { + t.Fatalf("expected temp file to be removed") + } +} + +func TestAgentCheckSilencePublishesInboundAndCleansUp(t *testing.T) { + t.Parallel() + + mb := bus.NewMessageBus() + defer mb.Close() + + tr := &fakeTranscriber{text: "hello there"} + agent := NewAgent(mb, tr) + + filePath := filepath.Join(t.TempDir(), "voice.ogg") + writer, err := oggwriter.New(filePath, 48000, 2) + if err != nil { + t.Fatalf("create ogg writer: %v", err) + } + + acc := &speechAccumulator{ + writer: writer, + file: filePath, + lastAudioAt: time.Now().Add(-2 * time.Second), + chatID: "chat", + speakerID: "speaker", + sessionID: "sess", + channel: "slack", + } + + agent.mu.Lock() + agent.sessions["sess_speaker"] = acc + agent.mu.Unlock() + + agent.checkSilence(context.Background()) + + select { + case msg := <-mb.InboundChan(): + if msg.Channel != "slack" { + t.Fatalf("unexpected inbound channel: %q", msg.Channel) + } + if !strings.Contains(msg.Content, "hello there") { + t.Fatalf("unexpected inbound content: %q", msg.Content) + } + if msg.Metadata["is_voice"] != "true" { + t.Fatalf("expected is_voice metadata, got %#v", msg.Metadata) + } + case <-time.After(500 * time.Millisecond): + t.Fatal("expected inbound publish") + } + + waitForFileRemoval(t, filePath, 500*time.Millisecond) +} diff --git a/pkg/voice/transcriber.go b/pkg/audio/asr/asr.go similarity index 93% rename from pkg/voice/transcriber.go rename to pkg/audio/asr/asr.go index f56fdeedd..1c4bbbb8f 100644 --- a/pkg/voice/transcriber.go +++ b/pkg/audio/asr/asr.go @@ -1,4 +1,4 @@ -package voice +package asr import ( "context" @@ -60,7 +60,8 @@ func DetectTranscriber(cfg *config.Config) Transcriber { } // Fall back to any model-list entry that uses the groq/ protocol. for _, mc := range cfg.ModelList { - if strings.HasPrefix(mc.Model, "groq/") && mc.APIKey() != "" { + if (strings.HasPrefix(mc.Model, "groq/") || mc.ModelName == "groq" || mc.Model == "whisper-large-v3-turbo") && + mc.APIKey() != "" { return NewGroqTranscriber(mc.APIKey()) } } diff --git a/pkg/voice/transcriber_test.go b/pkg/audio/asr/asr_test.go similarity index 99% rename from pkg/voice/transcriber_test.go rename to pkg/audio/asr/asr_test.go index 3e71ff13a..4935c9f85 100644 --- a/pkg/voice/transcriber_test.go +++ b/pkg/audio/asr/asr_test.go @@ -1,4 +1,4 @@ -package voice +package asr import ( "testing" diff --git a/pkg/voice/audio_model_transcriber.go b/pkg/audio/asr/audio_model_transcriber.go similarity index 99% rename from pkg/voice/audio_model_transcriber.go rename to pkg/audio/asr/audio_model_transcriber.go index f3ca81961..e8ded15dd 100644 --- a/pkg/voice/audio_model_transcriber.go +++ b/pkg/audio/asr/audio_model_transcriber.go @@ -1,4 +1,4 @@ -package voice +package asr import ( "context" diff --git a/pkg/voice/audio_model_transcriber_test.go b/pkg/audio/asr/audio_model_transcriber_test.go similarity index 99% rename from pkg/voice/audio_model_transcriber_test.go rename to pkg/audio/asr/audio_model_transcriber_test.go index c33e3bf97..5aaa82061 100644 --- a/pkg/voice/audio_model_transcriber_test.go +++ b/pkg/audio/asr/audio_model_transcriber_test.go @@ -1,4 +1,4 @@ -package voice +package asr import ( "context" diff --git a/pkg/voice/elevenlabs_transcriber.go b/pkg/audio/asr/elevenlabs_transcriber.go similarity index 99% rename from pkg/voice/elevenlabs_transcriber.go rename to pkg/audio/asr/elevenlabs_transcriber.go index 93db10f8d..d94522314 100644 --- a/pkg/voice/elevenlabs_transcriber.go +++ b/pkg/audio/asr/elevenlabs_transcriber.go @@ -1,4 +1,4 @@ -package voice +package asr import ( "bytes" diff --git a/pkg/voice/elevenlabs_transcriber_test.go b/pkg/audio/asr/elevenlabs_transcriber_test.go similarity index 99% rename from pkg/voice/elevenlabs_transcriber_test.go rename to pkg/audio/asr/elevenlabs_transcriber_test.go index 78be8958a..fb28cdd60 100644 --- a/pkg/voice/elevenlabs_transcriber_test.go +++ b/pkg/audio/asr/elevenlabs_transcriber_test.go @@ -1,4 +1,4 @@ -package voice +package asr import ( "context" diff --git a/pkg/voice/groq_transcriber.go b/pkg/audio/asr/groq_transcriber.go similarity index 67% rename from pkg/voice/groq_transcriber.go rename to pkg/audio/asr/groq_transcriber.go index b42e598f7..46c6201f9 100644 --- a/pkg/voice/groq_transcriber.go +++ b/pkg/audio/asr/groq_transcriber.go @@ -1,4 +1,4 @@ -package voice +package asr import ( "bytes" @@ -35,45 +35,28 @@ func NewGroqTranscriber(apiKey string) *GroqTranscriber { } } -func (t *GroqTranscriber) Transcribe(ctx context.Context, audioFilePath string) (*TranscriptionResponse, error) { - logger.InfoCF("voice", "Starting transcription", map[string]any{"audio_file": audioFilePath}) - - audioFile, err := os.Open(audioFilePath) - if err != nil { - logger.ErrorCF("voice", "Failed to open audio file", map[string]any{"path": audioFilePath, "error": err}) - return nil, fmt.Errorf("failed to open audio file: %w", err) - } - defer audioFile.Close() - - fileInfo, err := audioFile.Stat() - if err != nil { - logger.ErrorCF("voice", "Failed to get file info", map[string]any{"path": audioFilePath, "error": err}) - return nil, fmt.Errorf("failed to get file info: %w", err) - } - - logger.DebugCF("voice", "Audio file details", map[string]any{ - "size_bytes": fileInfo.Size(), - "file_name": filepath.Base(audioFilePath), - }) +func (t *GroqTranscriber) TranscribeData( + ctx context.Context, + data []byte, + filename string, +) (*TranscriptionResponse, error) { + logger.InfoCF("voice", "Starting memory transcription", map[string]any{"filename": filename, "bytes": len(data)}) var requestBody bytes.Buffer writer := multipart.NewWriter(&requestBody) - part, err := writer.CreateFormFile("file", filepath.Base(audioFilePath)) + part, err := writer.CreateFormFile("file", filename) if err != nil { logger.ErrorCF("voice", "Failed to create form file", map[string]any{"error": err}) return nil, fmt.Errorf("failed to create form file: %w", err) } - copied, err := io.Copy(part, audioFile) - if err != nil { - logger.ErrorCF("voice", "Failed to copy file content", map[string]any{"error": err}) - return nil, fmt.Errorf("failed to copy file content: %w", err) + if _, copyErr := io.Copy(part, bytes.NewReader(data)); copyErr != nil { + logger.ErrorCF("voice", "Failed to copy file content", map[string]any{"error": copyErr}) + return nil, fmt.Errorf("failed to copy file content: %w", copyErr) } - logger.DebugCF("voice", "File copied to request", map[string]any{"bytes_copied": copied}) - - if err = writer.WriteField("model", "whisper-large-v3"); err != nil { + if err = writer.WriteField("model", "whisper-large-v3-turbo"); err != nil { logger.ErrorCF("voice", "Failed to write model field", map[string]any{"error": err}) return nil, fmt.Errorf("failed to write model field: %w", err) } @@ -88,20 +71,70 @@ func (t *GroqTranscriber) Transcribe(ctx context.Context, audioFilePath string) return nil, fmt.Errorf("failed to close multipart writer: %w", err) } + return t.doRequest(ctx, &requestBody, writer.FormDataContentType(), int64(len(data))) +} + +func (t *GroqTranscriber) Transcribe(ctx context.Context, audioFilePath string) (*TranscriptionResponse, error) { + logger.InfoCF("voice", "Starting transcription", map[string]any{"audio_file": audioFilePath}) + + audioFile, err := os.Open(audioFilePath) + if err != nil { + 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, 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, 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) + } + + return t.doRequest(ctx, &requestBody, writer.FormDataContentType(), fileInfo.Size()) +} + +func (t *GroqTranscriber) doRequest( + ctx context.Context, + requestBody *bytes.Buffer, + contentType string, + fileSize int64, +) (*TranscriptionResponse, error) { url := t.apiBase + "/audio/transcriptions" - req, err := http.NewRequestWithContext(ctx, "POST", url, &requestBody) + req, err := http.NewRequestWithContext(ctx, "POST", url, requestBody) if err != nil { logger.ErrorCF("voice", "Failed to create request", map[string]any{"error": err}) return nil, fmt.Errorf("failed to create request: %w", err) } - req.Header.Set("Content-Type", writer.FormDataContentType()) + req.Header.Set("Content-Type", contentType) req.Header.Set("Authorization", "Bearer "+t.apiKey) logger.DebugCF("voice", "Sending transcription request to Groq API", map[string]any{ "url": url, "request_size_bytes": requestBody.Len(), - "file_size_bytes": fileInfo.Size(), + "file_size_bytes": fileSize, }) resp, err := t.httpClient.Do(req) diff --git a/pkg/audio/ogg.go b/pkg/audio/ogg.go new file mode 100644 index 000000000..f0055a574 --- /dev/null +++ b/pkg/audio/ogg.go @@ -0,0 +1,57 @@ +package audio + +import ( + "bytes" + "fmt" + "io" +) + +// DecodeOggOpus reads an Ogg format stream and extracts individual Opus payloads. +// It calls onFrame for every complete Opus frame found in the stream. +func DecodeOggOpus(r io.Reader, onFrame func([]byte) error) error { + var packet bytes.Buffer + header := make([]byte, 27) + segment := make([]byte, 255) + + for { + 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 { + if _, err := io.ReadFull(r, segment[:lacing]); err != nil { + return fmt.Errorf("failed to read segment data: %w", err) + } + + packet.Write(segment[:lacing]) + + // If lacing is less than 255, the packet is complete + if lacing < 255 { + if packet.Len() > 0 { + packetBytes := packet.Bytes() + // Ignore Ogg Opus headers + if !bytes.HasPrefix(packetBytes, []byte("OpusHead")) && + !bytes.HasPrefix(packetBytes, []byte("OpusTags")) { + if err := onFrame(packetBytes); err != nil { + return err + } + } + // Start new packet + packet.Reset() + } + } + } + } +} diff --git a/pkg/audio/ogg_test.go b/pkg/audio/ogg_test.go new file mode 100644 index 000000000..8d5e5ac2a --- /dev/null +++ b/pkg/audio/ogg_test.go @@ -0,0 +1,146 @@ +package audio + +import ( + "bytes" + "reflect" + "strings" + "testing" +) + +// buildOggPage helper creates an Ogg page for testing. +// lacingVals specifies the segment table, and data is the payload. +func buildOggPage(lacingVals []byte, data []byte) []byte { + var buf bytes.Buffer + // 27-byte Ogg header + header := make([]byte, 27) + copy(header[:4], "OggS") + header[5] = 0 // type flag + // For testing, we only care about OggS magic and page_segments (byte 26) + header[26] = byte(len(lacingVals)) + buf.Write(header) + buf.Write(lacingVals) + buf.Write(data) + return buf.Bytes() +} + +func TestDecodeOggOpus_ValidParsing(t *testing.T) { + var b bytes.Buffer + + // Packet 1: Single segment, length 50 + pkt1 := bytes.Repeat([]byte{1}, 50) + // Packet 2: Multi-segment (255 + 10 = 265 bytes) + pkt2Part1 := bytes.Repeat([]byte{2}, 255) + pkt2Part2 := bytes.Repeat([]byte{2}, 10) + // Packet 3: Continued across pages. Page 1 gets 255, Page 2 gets 20. Total 275 bytes. + pkt3Part1 := bytes.Repeat([]byte{3}, 255) + pkt3Part2 := bytes.Repeat([]byte{3}, 20) + + // Page 1: OpusHead (skip), OpusTags (skip), pkt1, pkt2, pkt3Part1 + page1Lacing := []byte{8, 8, 50, 255, 10, 255} + page1Data := bytes.Join([][]byte{ + []byte("OpusHead"), + []byte("OpusTags"), + pkt1, + pkt2Part1, pkt2Part2, + pkt3Part1, + }, nil) + + // Page 2: pkt3Part2, pkt4 (length 10) + pkt4 := bytes.Repeat([]byte{4}, 10) + page2Lacing := []byte{20, 10} + page2Data := bytes.Join([][]byte{ + pkt3Part2, + pkt4, + }, nil) + + b.Write(buildOggPage(page1Lacing, page1Data)) + b.Write(buildOggPage(page2Lacing, page2Data)) + + var frames [][]byte + err := DecodeOggOpus(&b, func(frame []byte) error { + // making a copy to store as DecodeOggOpus might reuse backing array + cpy := make([]byte, len(frame)) + copy(cpy, frame) + frames = append(frames, cpy) + return nil + }) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + expectedFrames := [][]byte{ + pkt1, + append(pkt2Part1, pkt2Part2...), + append(pkt3Part1, pkt3Part2...), + pkt4, + } + + if len(frames) != len(expectedFrames) { + t.Fatalf("expected %d frames, got %d", len(expectedFrames), len(frames)) + } + + for i, expected := range expectedFrames { + if !reflect.DeepEqual(frames[i], expected) { + t.Errorf("frame %d mismatch:\nexp: %v\ngot: %v", i, expected, frames[i]) + } + } +} + +func TestDecodeOggOpus_Errors(t *testing.T) { + tests := []struct { + name string + data []byte + errContains string + }{ + { + name: "invalid magic string", + data: []byte( + "OggX\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00", + ), + errContains: "invalid ogg magic string", + }, + { + name: "short header", + data: []byte("Ogg"), + errContains: "failed to read ogg header", + }, + { + name: "eof in segment table", + data: func() []byte { + h := make([]byte, 27) + copy(h, "OggS") + h[26] = 5 // expects 5 bytes of segment table, but none provided + return h + }(), + errContains: "failed to read segment table", + }, + { + name: "eof in segment data", + data: func() []byte { + h := make([]byte, 27, 28) + copy(h, "OggS") + h[26] = 1 + return append(h, 100) // expects 100 bytes of data, but none provided + }(), + errContains: "failed to read segment data", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + err := DecodeOggOpus(bytes.NewReader(tt.data), func(b []byte) error { return nil }) + if tt.name == "short header" { + if err != nil { + t.Errorf("expected no error (io.EOF/ErrUnexpectedEOF swallowed), got %v", err) + } + return + } + if err == nil { + t.Fatalf("expected error containing %q, got nil", tt.errContains) + } + if !strings.Contains(err.Error(), tt.errContains) { + t.Errorf("expected error to contain %q, got: %q", tt.errContains, err.Error()) + } + }) + } +} diff --git a/pkg/audio/sentence.go b/pkg/audio/sentence.go new file mode 100644 index 000000000..89b9ac03e --- /dev/null +++ b/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, as well as CJK 。, ！, ？) while avoiding false splits +// on 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] + if r == '\n' { + s := strings.TrimSpace(current.String()) + if s != "" { + sentences = append(sentences, s) + } + current.Reset() + continue + } + + current.WriteRune(r) + + if r == '.' || r == '!' || r == '?' || 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 contiguous punctuation clusters (e.g., "..." or "?!"). + for i+1 < len(runes) && (runes[i+1] == '.' || runes[i+1] == '!' || 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 +} diff --git a/pkg/audio/sentence_test.go b/pkg/audio/sentence_test.go new file mode 100644 index 000000000..54d69e4a6 --- /dev/null +++ b/pkg/audio/sentence_test.go @@ -0,0 +1,69 @@ +package audio + +import ( + "reflect" + "testing" +) + +func TestSplitSentences(t *testing.T) { + tests := []struct { + name string + in string + want []string + }{ + { + name: "empty input", + in: "", + want: nil, + }, + { + name: "single sentence", + in: "Hello world.", + want: []string{"Hello world."}, + }, + { + name: "decimal numbers do not split", + in: "The value is 3.14 today. Keep watching closely.", + want: []string{"The value is 3.14 today.", "Keep watching closely."}, + }, + { + name: "newline boundary", + in: "This is line number one\nThis is line number two", + want: []string{"This is line number one", "This is line number two"}, + }, + { + name: "newline with surrounding spaces", + in: " This is the first line \n This is the second line ", + want: []string{"This is the first line", "This is the second line"}, + }, + { + name: "trailing punctuation consumed", + in: "Please wait a moment... What on earth?! That is perfectly fine.", + want: []string{"Please wait a moment...", "What on earth?!", "That is perfectly fine."}, + }, + { + name: "short leading fragment merges with next", + in: "Hi. This is a longer sentence.", + want: []string{"Hi. This is a longer sentence."}, + }, + { + name: "consecutive short fragments keep merging", + in: "A. B. C. This is the real sentence.", + want: []string{"A. B. C. This is the real sentence."}, + }, + { + name: "short trailing fragment merges back", + in: "This sentence is long enough. End.", + want: []string{"This sentence is long enough. End."}, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + got := SplitSentences(tc.in) + if !reflect.DeepEqual(got, tc.want) { + t.Fatalf("SplitSentences(%q) = %#v, want %#v", tc.in, got, tc.want) + } + }) + } +} diff --git a/pkg/audio/tts/openai_tts.go b/pkg/audio/tts/openai_tts.go new file mode 100644 index 000000000..792a6f220 --- /dev/null +++ b/pkg/audio/tts/openai_tts.go @@ -0,0 +1,121 @@ +package tts + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "net/url" + "strings" + "time" + + "github.com/sipeed/picoclaw/pkg/logger" + "github.com/sipeed/picoclaw/pkg/providers/common" +) + +type OpenAITTSProvider struct { + apiKey string + apiBase string + voice string + model string + httpClient *http.Client +} + +func NewOpenAITTSProvider(apiKey string, apiBase string, proxyURL string) *OpenAITTSProvider { + // Normalize apiBase to avoid malformed endpoints like + // "https://api.openai.com/audio/speech" when "/v1" is required. + if apiBase == "" { + apiBase = "https://api.openai.com/v1/audio/speech" + } else { + if u, err := url.Parse(apiBase); err == nil && u.Scheme != "" && u.Host != "" { + path := u.Path + if u.Host == "api.openai.com" { + // For the official OpenAI host, ensure exactly one /v1 prefix and + // that the path ends with /audio/speech. + if path == "" || path == "/" || path == "/v1" { + path = "/v1/audio/speech" + } else { + if !strings.HasPrefix(path, "/") { + path = "/" + path + } + if !strings.HasPrefix(path, "/v1/") { + path = "/v1" + strings.TrimSuffix(path, "/") + } + if !strings.HasSuffix(path, "/audio/speech") { + path = strings.TrimSuffix(path, "/") + "/audio/speech" + } + } + } else { + // For non-OpenAI hosts (e.g., proxies), preserve the existing base + // path and only ensure it ends with /audio/speech. + if !strings.HasSuffix(path, "/audio/speech") { + path = strings.TrimSuffix(path, "/") + "/audio/speech" + } + } + u.Path = path + apiBase = u.String() + } else { + // Fallback to the previous string-based behavior if parsing fails. + if apiBase == "https://api.openai.com/v1" { + apiBase = "https://api.openai.com/v1/audio/speech" + } else if !strings.HasSuffix(apiBase, "/audio/speech") { + // Just in case they provide openrouter base or standard base + apiBase = strings.TrimSuffix(apiBase, "/") + "/audio/speech" + } + } + } + + client := common.NewHTTPClient(proxyURL) + client.Timeout = 60 * time.Second + + return &OpenAITTSProvider{ + apiKey: apiKey, + apiBase: apiBase, + voice: "alloy", + model: "tts-1", + httpClient: client, + } +} + +func (t *OpenAITTSProvider) Name() string { + return "openai-tts" +} + +func (t *OpenAITTSProvider) Synthesize(ctx context.Context, text string) (io.ReadCloser, error) { + logger.DebugCF("voice-tts", "Starting TTS synthesis", map[string]any{"text_len": len(text)}) + + reqBody := map[string]any{ + "model": t.model, + "input": text, + "voice": t.voice, + "response_format": "opus", + } + + jsonData, err := json.Marshal(reqBody) + if err != nil { + return nil, fmt.Errorf("failed to marshal request: %w", err) + } + + req, err := http.NewRequestWithContext(ctx, "POST", t.apiBase, bytes.NewReader(jsonData)) + if err != nil { + return nil, fmt.Errorf("failed to create request: %w", err) + } + + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Authorization", "Bearer "+t.apiKey) + + resp, err := t.httpClient.Do(req) + if err != nil { + return nil, fmt.Errorf("failed to send request: %w", err) + } + + if resp.StatusCode != http.StatusOK { + defer resp.Body.Close() + body, _ := io.ReadAll(resp.Body) + return nil, fmt.Errorf("API error (status %d): %s", resp.StatusCode, string(body)) + } + + return resp.Body, nil +} diff --git a/pkg/audio/tts/tts.go b/pkg/audio/tts/tts.go new file mode 100644 index 000000000..e1c4fcda0 --- /dev/null +++ b/pkg/audio/tts/tts.go @@ -0,0 +1,111 @@ +package tts + +import ( + "context" + "fmt" + "io" + "os" + "path/filepath" + "strings" + "time" + + "github.com/sipeed/picoclaw/pkg/config" + "github.com/sipeed/picoclaw/pkg/media" +) + +type TTSProvider interface { + Name() string + Synthesize(ctx context.Context, text string) (io.ReadCloser, error) +} + +func DetectTTS(cfg *config.Config) TTSProvider { + for _, mc := range cfg.ModelList { + if strings.Contains(strings.ToLower(mc.Model), "tts") && mc.APIKey() != "" { + return NewOpenAITTSProvider(mc.APIKey(), mc.APIBase, mc.Proxy) + } + } + return nil +} + +// SynthesizeAndStore synthesizes text to speech and registers it in the media store, returning the media reference. +func SynthesizeAndStore( + ctx context.Context, + provider TTSProvider, + store media.MediaStore, + text string, + filename string, + channel string, + chatID string, +) (string, error) { + if provider == nil { + return "", fmt.Errorf("tts provider is not configured") + } + if store == nil { + return "", fmt.Errorf("media store not configured") + } + if channel == "" || chatID == "" { + return "", fmt.Errorf("no target channel/chat available") + } + if strings.TrimSpace(text) == "" { + return "", fmt.Errorf("text is required") + } + + stream, err := provider.Synthesize(ctx, text) + if err != nil { + return "", fmt.Errorf("tts synthesize failed: %w", err) + } + defer stream.Close() + + err = os.MkdirAll(media.TempDir(), 0o700) + if err != nil { + return "", fmt.Errorf("failed to create media temp dir: %w", err) + } + + file, err := os.CreateTemp(media.TempDir(), "tts-*.ogg") + if err != nil { + return "", fmt.Errorf("failed to create temp file: %w", err) + } + + removeTemp := true + defer func() { + if removeTemp { + _ = os.Remove(file.Name()) + } + }() + + _, err = io.Copy(file, stream) + if err != nil { + file.Close() + return "", fmt.Errorf("failed to write tts audio: %w", err) + } + + err = file.Close() + if err != nil { + return "", fmt.Errorf("failed to close tts audio file: %w", err) + } + + filename = strings.TrimSpace(filename) + if filename == "" { + filename = fmt.Sprintf("tts-%d.ogg", time.Now().Unix()) + } + + ext := strings.ToLower(filepath.Ext(filename)) + if ext == "" { + filename += ".ogg" + } else if ext != ".ogg" { + filename = strings.TrimSuffix(filename, filepath.Ext(filename)) + ".ogg" + } + + scope := fmt.Sprintf("tool:send_tts:%s:%s:%d", channel, chatID, time.Now().UnixNano()) + ref, err := store.Store(file.Name(), media.MediaMeta{ + Filename: filename, + ContentType: "audio/ogg", + Source: "tool:send_tts", + }, scope) + if err != nil { + return "", fmt.Errorf("failed to register audio: %w", err) + } + removeTemp = false + + return ref, nil +} diff --git a/pkg/audio/tts/tts_test.go b/pkg/audio/tts/tts_test.go new file mode 100644 index 000000000..87409ff65 --- /dev/null +++ b/pkg/audio/tts/tts_test.go @@ -0,0 +1,129 @@ +package tts + +import ( + "context" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" +) + +func TestNewOpenAITTSProvider_APIBaseNormalization(t *testing.T) { + t.Parallel() + + cases := []struct { + name string + input string + expect string + }{ + { + name: "empty base", + input: "", + expect: "https://api.openai.com/v1/audio/speech", + }, + { + name: "official host no path", + input: "https://api.openai.com", + expect: "https://api.openai.com/v1/audio/speech", + }, + { + name: "official host v1", + input: "https://api.openai.com/v1", + expect: "https://api.openai.com/v1/audio/speech", + }, + { + name: "official host v1 slash", + input: "https://api.openai.com/v1/", + expect: "https://api.openai.com/v1/audio/speech", + }, + { + name: "non-openai host preserves base path", + input: "https://proxy.example.com/base", + expect: "https://proxy.example.com/base/audio/speech", + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + provider := NewOpenAITTSProvider("key", tc.input, "") + if provider.apiBase != tc.expect { + t.Fatalf("apiBase mismatch: got %q, want %q", provider.apiBase, tc.expect) + } + }) + } +} + +func TestOpenAITTSProvider_SynthesizeSuccess(t *testing.T) { + t.Parallel() + + var gotPath string + var gotAuth string + var gotContentType string + var gotBody map[string]any + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotPath = r.URL.Path + gotAuth = r.Header.Get("Authorization") + gotContentType = r.Header.Get("Content-Type") + + bodyBytes, _ := io.ReadAll(r.Body) + _ = r.Body.Close() + _ = json.Unmarshal(bodyBytes, &gotBody) + + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte("audio-bytes")) + })) + defer server.Close() + + provider := NewOpenAITTSProvider("k123", server.URL, "") + stream, err := provider.Synthesize(context.Background(), "hello") + if err != nil { + t.Fatalf("Synthesize failed: %v", err) + } + defer stream.Close() + + data, err := io.ReadAll(stream) + if err != nil { + t.Fatalf("read stream failed: %v", err) + } + + if gotPath != "/audio/speech" { + t.Fatalf("request path mismatch: got %q", gotPath) + } + if gotAuth != "Bearer k123" { + t.Fatalf("authorization mismatch: got %q", gotAuth) + } + if gotContentType != "application/json" { + t.Fatalf("content-type mismatch: got %q", gotContentType) + } + if gotBody["model"] != "tts-1" || gotBody["voice"] != "alloy" || gotBody["response_format"] != "opus" || + gotBody["input"] != "hello" { + bodyJSON, _ := json.Marshal(gotBody) + t.Fatalf("request body mismatch: %s", string(bodyJSON)) + } + if string(data) != "audio-bytes" { + t.Fatalf("response body mismatch: got %q", string(data)) + } +} + +func TestOpenAITTSProvider_SynthesizeNon200(t *testing.T) { + t.Parallel() + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusInternalServerError) + _, _ = w.Write([]byte("nope")) + })) + defer server.Close() + + provider := NewOpenAITTSProvider("k123", server.URL, "") + _, err := provider.Synthesize(context.Background(), "hello") + if err == nil { + t.Fatal("expected error") + } + if !strings.Contains(err.Error(), "API error (status 500): nope") { + t.Fatalf("unexpected error: %v", err) + } +} diff --git a/pkg/bus/bus.go b/pkg/bus/bus.go index 37fcb74c5..a9c74ef90 100644 --- a/pkg/bus/bus.go +++ b/pkg/bus/bus.go @@ -34,6 +34,8 @@ type MessageBus struct { inbound chan InboundMessage outbound chan OutboundMessage outboundMedia chan OutboundMediaMessage + audioChunks chan AudioChunk + voiceControls chan VoiceControl closeOnce sync.Once done chan struct{} @@ -47,6 +49,8 @@ func NewMessageBus() *MessageBus { inbound: make(chan InboundMessage, defaultBusBufferSize), outbound: make(chan OutboundMessage, defaultBusBufferSize), outboundMedia: make(chan OutboundMediaMessage, defaultBusBufferSize), + audioChunks: make(chan AudioChunk, defaultBusBufferSize*4), // Audio chunks need more buffer + voiceControls: make(chan VoiceControl, defaultBusBufferSize), done: make(chan struct{}), } } @@ -103,6 +107,22 @@ func (mb *MessageBus) OutboundMediaChan() <-chan OutboundMediaMessage { return mb.outboundMedia } +func (mb *MessageBus) PublishAudioChunk(ctx context.Context, chunk AudioChunk) error { + return publish(ctx, mb, mb.audioChunks, chunk) +} + +func (mb *MessageBus) AudioChunksChan() <-chan AudioChunk { + return mb.audioChunks +} + +func (mb *MessageBus) PublishVoiceControl(ctx context.Context, ctrl VoiceControl) error { + return publish(ctx, mb, mb.voiceControls, ctrl) +} + +func (mb *MessageBus) VoiceControlsChan() <-chan VoiceControl { + return mb.voiceControls +} + // SetStreamDelegate registers a StreamDelegate (typically the channel Manager). func (mb *MessageBus) SetStreamDelegate(d StreamDelegate) { mb.streamDelegate.Store(d) @@ -132,6 +152,8 @@ func (mb *MessageBus) Close() { close(mb.inbound) close(mb.outbound) close(mb.outboundMedia) + close(mb.audioChunks) + close(mb.voiceControls) // clean up any remaining messages in channels drained := 0 @@ -144,6 +166,12 @@ func (mb *MessageBus) Close() { for range mb.outboundMedia { drained++ } + for range mb.audioChunks { + drained++ + } + for range mb.voiceControls { + drained++ + } if drained > 0 { logger.DebugCF("bus", "Drained buffered messages during close", map[string]any{ diff --git a/pkg/bus/types.go b/pkg/bus/types.go index 12da3f1dd..27cf61b5f 100644 --- a/pkg/bus/types.go +++ b/pkg/bus/types.go @@ -30,10 +30,11 @@ type InboundMessage struct { } type OutboundMessage struct { - Channel string `json:"channel"` - ChatID string `json:"chat_id"` - Content string `json:"content"` - ReplyToMessageID string `json:"reply_to_message_id,omitempty"` + Channel string `json:"channel"` + ChatID string `json:"chat_id"` + Content string `json:"content"` + ReplyToMessageID string `json:"reply_to_message_id,omitempty"` + Metadata map[string]string `json:"metadata,omitempty"` } // MediaPart describes a single media attachment to send. @@ -51,3 +52,25 @@ type OutboundMediaMessage struct { ChatID string `json:"chat_id"` Parts []MediaPart `json:"parts"` } + +// AudioChunk represents a chunk of streaming voice data. +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"` + Channels int `json:"channels"` + Format string `json:"format"` // "opus", "pcm", etc + Data []byte `json:"data"` +} + +// VoiceControl represents state or commands for voice sessions. +type VoiceControl struct { + SessionID string `json:"session_id"` + ChatID string `json:"chat_id"` + Type string `json:"type"` // "state", "command" + Action string `json:"action"` // "idle", "listening", "start", "stop", "leave" +} diff --git a/pkg/channels/discord/discord.go b/pkg/channels/discord/discord.go index cc0ef4ffe..42024b270 100644 --- a/pkg/channels/discord/discord.go +++ b/pkg/channels/discord/discord.go @@ -3,6 +3,7 @@ package discord import ( "context" "fmt" + "io" "net/http" "net/url" "os" @@ -14,6 +15,8 @@ import ( "github.com/bwmarrin/discordgo" "github.com/gorilla/websocket" + "github.com/sipeed/picoclaw/pkg/audio" + "github.com/sipeed/picoclaw/pkg/audio/tts" "github.com/sipeed/picoclaw/pkg/bus" "github.com/sipeed/picoclaw/pkg/channels" "github.com/sipeed/picoclaw/pkg/config" @@ -42,6 +45,15 @@ type DiscordChannel struct { typingMu sync.Mutex typingStop map[string]chan struct{} // chatID → stop signal 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) { @@ -73,6 +85,8 @@ func NewDiscordChannel(cfg config.DiscordConfig, bus *bus.MessageBus) (*DiscordC config: cfg, ctx: context.Background(), typingStop: make(map[string]chan struct{}), + bus: bus, + voiceSSRC: make(map[string]map[uint32]string), }, nil } @@ -90,6 +104,8 @@ func (c *DiscordChannel) Start(ctx context.Context) error { c.session.AddHandler(c.handleMessage) + go c.listenVoiceControl(c.ctx) + if err := c.session.Open(); err != nil { return fmt.Errorf("failed to open discord session: %w", err) } @@ -142,6 +158,25 @@ func (c *DiscordChannel) Send(ctx context.Context, msg bus.OutboundMessage) erro return nil } + 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 + c.ttsMu.Lock() + if c.cancelTTS != nil { + 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, playID) + } + } + } + return c.sendChunk(ctx, channelID, msg.Content, msg.ReplyToMessageID) } @@ -339,6 +374,10 @@ func (c *DiscordChannel) handleMessage(s *discordgo.Session, m *discordgo.Messag return } + if c.handleVoiceCommand(s, m) { + return + } + content := m.Content // In guild (group) channels, apply unified group trigger filtering @@ -610,3 +649,134 @@ func (c *DiscordChannel) stripBotMention(text string) string { text = strings.ReplaceAll(text, fmt.Sprintf("<@!%s>", c.botUserID), "") return strings.TrimSpace(text) } + +func (c *DiscordChannel) listenVoiceControl(ctx context.Context) { + for { + select { + case <-ctx.Done(): + return + 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_") + vc, exists := c.session.VoiceConnections[guildID] + if exists && vc != nil { + vc.Disconnect(ctx) + } + } + } + } + } +} + +func (c *DiscordChannel) playTTS(ctx context.Context, vc *discordgo.VoiceConnection, text string, playID uint64) { + // Capture the cancel func associated with this playback (if any). + // 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.ttsPlayID == playID { + c.cancelTTS = nil + } + c.ttsMu.Unlock() + }() + + sentences := audio.SplitSentences(text) + if len(sentences) == 0 { + return + } + + logger.InfoCF("discord", "Starting streamed TTS", map[string]any{"sentences": len(sentences)}) + + // Pipeline: prefetch next sentence's audio while playing current + type ttResult struct { + stream io.ReadCloser + err error + } + + var prefetch chan ttResult + + // Ensure any in-flight prefetch is drained on exit to prevent stream leaks, + // but avoid blocking indefinitely if the prefetch goroutine is stuck or never sends. + defer func() { + if prefetch != nil { + select { + case result := <-prefetch: + if result.stream != nil { + result.stream.Close() + } + case <-time.After(100 * time.Millisecond): + // Timed out waiting for a prefetched result; avoid blocking on exit. + } + } + }() + + 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, but be responsive to cancellation. + var result ttResult + select { + case result = <-prefetch: + stream, err = result.stream, result.err + case <-ctx.Done(): + // Context canceled while waiting for prefetched audio; abort playback. + logger.InfoCF( + "discord", + "TTS interrupted while waiting for prefetched audio", + map[string]any{"at_sentence": i}, + ) + return + } + } else { + // First sentence: synthesize directly + stream, err = c.tts.Synthesize(ctx, sentence) + } + + if err != nil { + if stream != nil { + stream.Close() + } + 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 + } +} + +// VoiceCapabilities returns the voice capabilities of the channel. +func (c *DiscordChannel) VoiceCapabilities() channels.VoiceCapabilities { + return channels.VoiceCapabilities{ASR: true, TTS: true} +} diff --git a/pkg/channels/discord/init.go b/pkg/channels/discord/init.go index 15a539804..8381dc9e9 100644 --- a/pkg/channels/discord/init.go +++ b/pkg/channels/discord/init.go @@ -1,6 +1,7 @@ package discord import ( + "github.com/sipeed/picoclaw/pkg/audio/tts" "github.com/sipeed/picoclaw/pkg/bus" "github.com/sipeed/picoclaw/pkg/channels" "github.com/sipeed/picoclaw/pkg/config" @@ -8,6 +9,10 @@ import ( func init() { channels.RegisterFactory("discord", func(cfg *config.Config, b *bus.MessageBus) (channels.Channel, error) { - return NewDiscordChannel(cfg.Channels.Discord, b) + ch, err := NewDiscordChannel(cfg.Channels.Discord, b) + if err == nil { + ch.tts = tts.DetectTTS(cfg) + } + return ch, err }) } diff --git a/pkg/channels/discord/voice.go b/pkg/channels/discord/voice.go new file mode 100644 index 000000000..5b686b141 --- /dev/null +++ b/pkg/channels/discord/voice.go @@ -0,0 +1,313 @@ +package discord + +import ( + "context" + "fmt" + "io" + "time" + + "github.com/bwmarrin/discordgo" + + "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 { + logger.InfoCF("discord", "Failed to send voice channel requirement message", map[string]any{ + "channel": m.ChannelID, + "error": sendErr, + }) + } + return true + } + + logger.InfoCF("discord", "Joining voice channel", map[string]any{"channel": vs.ChannelID}) + 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 { + logger.InfoCF("discord", "Failed to send voice join error message", map[string]any{ + "channel": m.ChannelID, + "error": sendErr, + }) + } + return true + } + + go c.receiveVoice(vc, m.GuildID, m.ChannelID) + 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, + }) + } + return true + } else if m.Content == "!vc leave" { + vc, exists := s.VoiceConnections[m.GuildID] + if exists && vc != nil { + if err := vc.Disconnect(c.ctx); err != nil { + logger.InfoCF("discord", "Failed to disconnect from voice channel", map[string]any{ + "guild": m.GuildID, + "error": err, + }) + } + if _, sendErr := s.ChannelMessageSend(m.ChannelID, "Left Voice Channel."); sendErr != nil { + logger.InfoCF("discord", "Failed to send voice leave success message", map[string]any{ + "channel": m.ChannelID, + "error": sendErr, + }) + } + } else { + if _, sendErr := s.ChannelMessageSend(m.ChannelID, "Not in a voice channel."); sendErr != nil { + logger.InfoCF("discord", "Failed to send voice not-in-channel message", map[string]any{ + "channel": m.ChannelID, + "error": sendErr, + }) + } + } + return true + } + return false +} + +func VoiceReceiveActive(vc *discordgo.VoiceConnection) bool { + return vc != nil && vc.OpusRecv != nil +} + +func streamOggOpusToDiscord(ctx context.Context, vc *discordgo.VoiceConnection, r io.Reader) (retErr error) { + // Recover from panic if vc.OpusSend is closed mid-send (e.g. on disconnect) + defer func() { + if rec := recover(); rec != nil { + retErr = fmt.Errorf("voice connection closed during playback") + } + }() + + // Wait for the speaking transition to register + vc.Speaking(true) + defer vc.Speaking(false) + + return audio.DecodeOggOpus(r, func(frame []byte) error { + select { + case <-ctx.Done(): + return ctx.Err() + case vc.OpusSend <- frame: + return nil + } + }) +} + +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() { + if rec := recover(); rec != nil { + logger.WarnCF("discord", "Recovered from panic while sending wake-up frames", map[string]any{ + "error": rec, + "guild": guildID, + }) + } + }() + + // If the voice connection or OpusSend are not available, nothing to do. + if vc == nil || vc.OpusSend == nil { + return + } + + time.Sleep(250 * time.Millisecond) // Wait a bit for connection to settle + + // Abort if the context has already been canceled. + select { + case <-ctx.Done(): + return + default: + } + + vc.Speaking(true) + defer vc.Speaking(false) + + silenceFrame := []byte{0xF8, 0xFF, 0xFE} + for i := 0; i < 5; i++ { + select { + case <-ctx.Done(): + return + case vc.OpusSend <- silenceFrame: + } + time.Sleep(20 * time.Millisecond) + } + + logger.DebugCF("discord", "Sent wake-up silence frames", map[string]any{"guild": guildID}) + }(c.ctx, vc) + sessionID := fmt.Sprintf("discord_vc_%s", guildID) + + c.bus.PublishVoiceControl(c.ctx, bus.VoiceControl{ + SessionID: sessionID, + Type: "state", + Action: "listening", + }) + + var sequence uint64 = 0 + var interruptCount int + var lastInterruptAt time.Time + + for { + select { + case <-c.ctx.Done(): + return + case p, ok := <-vc.OpusRecv: + if !ok { + logger.InfoCF("discord", "Voice channel closed", map[string]any{"guild": guildID}) + // Cancel any TTS that may still be playing + c.ttsMu.Lock() + if c.cancelTTS != nil { + c.cancelTTS() + c.cancelTTS = nil + } + c.ttsMu.Unlock() + return + } + + if p == nil { + logger.DebugCF("discord", "Received nil Opus packet", nil) + continue + } + + if len(p.Opus) == 0 { + logger.DebugCF("discord", "Received empty Opus packet", map[string]any{ + "seq": p.Sequence, + "ssrc": p.SSRC, + }) + continue + } + + logger.DebugCF("discord", "Received Opus packet", map[string]any{ + "seq": p.Sequence, + "len": len(p.Opus), + "ssrc": p.SSRC, + }) + // 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 + } + + 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: userID, + ChatID: chatID, + Channel: "discord", + Sequence: sequence, + Timestamp: p.Timestamp, + SampleRate: 48000, + Channels: 2, + Format: "opus", + Data: p.Opus, + } + + ctx, cancel := context.WithTimeout(c.ctx, 100*time.Millisecond) + err := c.bus.PublishAudioChunk(ctx, chunk) + cancel() + if err != nil { + logger.ErrorCF("discord", "Failed to publish audio chunk", map[string]any{ + "guild": guildID, + "sessionID": sessionID, + "sequence": sequence, + "error": err.Error(), + }) + } + } + } +} diff --git a/pkg/channels/feishu/common.go b/pkg/channels/feishu/common.go index 4952394b7..81238460a 100644 --- a/pkg/channels/feishu/common.go +++ b/pkg/channels/feishu/common.go @@ -6,6 +6,8 @@ import ( "strings" larkim "github.com/larksuite/oapi-sdk-go/v3/service/im/v1" + + "github.com/sipeed/picoclaw/pkg/channels" ) // mentionPlaceholderRegex matches @_user_N placeholders inserted by Feishu for mentions. @@ -145,3 +147,8 @@ func extractImageKeysRecursive(v any, feishuKeys, externalURLs *[]string) { } } } + +// VoiceCapabilities returns the voice capabilities of the channel. +func (c *FeishuChannel) VoiceCapabilities() channels.VoiceCapabilities { + return channels.VoiceCapabilities{ASR: true, TTS: true} +} diff --git a/pkg/channels/line/line.go b/pkg/channels/line/line.go index 867ab24ee..dd5f2d25f 100644 --- a/pkg/channels/line/line.go +++ b/pkg/channels/line/line.go @@ -684,3 +684,8 @@ func (c *LINEChannel) downloadContent(messageID, filename string) string { }, }) } + +// VoiceCapabilities returns the voice capabilities of the channel. +func (c *LINEChannel) VoiceCapabilities() channels.VoiceCapabilities { + return channels.VoiceCapabilities{ASR: true, TTS: true} +} diff --git a/pkg/channels/matrix/matrix.go b/pkg/channels/matrix/matrix.go index 09b4eaa76..64407c1fc 100644 --- a/pkg/channels/matrix/matrix.go +++ b/pkg/channels/matrix/matrix.go @@ -1294,3 +1294,8 @@ func stripUserMentionWithRegexp(text string, userID id.UserID, mentionR *regexp. cleaned = strings.TrimLeft(cleaned, ",:; ") return strings.TrimSpace(cleaned) } + +// VoiceCapabilities returns the voice capabilities of the channel. +func (c *MatrixChannel) VoiceCapabilities() channels.VoiceCapabilities { + return channels.VoiceCapabilities{ASR: true, TTS: true} +} diff --git a/pkg/channels/onebot/onebot.go b/pkg/channels/onebot/onebot.go index e5f8b8fd1..2cbc5bc72 100644 --- a/pkg/channels/onebot/onebot.go +++ b/pkg/channels/onebot/onebot.go @@ -1104,3 +1104,8 @@ func truncate(s string, n int) string { } return string(runes[:n]) + "..." } + +// VoiceCapabilities returns the voice capabilities of the channel. +func (c *OneBotChannel) VoiceCapabilities() channels.VoiceCapabilities { + return channels.VoiceCapabilities{ASR: true, TTS: true} +} diff --git a/pkg/channels/qq/qq.go b/pkg/channels/qq/qq.go index dfea85ba3..0f60c2a6f 100644 --- a/pkg/channels/qq/qq.go +++ b/pkg/channels/qq/qq.go @@ -991,3 +991,8 @@ func sanitizeURLs(text string) string { return scheme + domain + path }) } + +// VoiceCapabilities returns the voice capabilities of the channel. +func (c *QQChannel) VoiceCapabilities() channels.VoiceCapabilities { + return channels.VoiceCapabilities{ASR: true, TTS: true} +} diff --git a/pkg/channels/telegram/telegram.go b/pkg/channels/telegram/telegram.go index f64a8f79b..3e68c10d1 100644 --- a/pkg/channels/telegram/telegram.go +++ b/pkg/channels/telegram/telegram.go @@ -985,3 +985,8 @@ func cryptoRandInt() int { _, _ = rand.Read(b[:]) return int(binary.BigEndian.Uint32(b[:])) | 1 // ensure non-zero } + +// VoiceCapabilities returns the voice capabilities of the channel. +func (c *TelegramChannel) VoiceCapabilities() channels.VoiceCapabilities { + return channels.VoiceCapabilities{ASR: true, TTS: true} +} diff --git a/pkg/channels/voice_capabilities.go b/pkg/channels/voice_capabilities.go new file mode 100644 index 000000000..34fd24269 --- /dev/null +++ b/pkg/channels/voice_capabilities.go @@ -0,0 +1,58 @@ +package channels + +// VoiceCapabilities describes whether ASR (speech-to-text) and TTS (text-to-speech) +// are available for a channel under the current configuration. +type VoiceCapabilities struct { + ASR bool + TTS bool +} + +// VoiceCapabilityProvider is an optional interface for channels that want to +// explicitly declare their ASR/TTS support. +type VoiceCapabilityProvider interface { + VoiceCapabilities() VoiceCapabilities +} + +// Deprecated: Channels should implement VoiceCapabilityProvider instead. +// To be removed once all existing capable channels conform to the interface. +var asrCapableChannels = map[string]bool{ + "discord": true, + "telegram": true, + "matrix": true, + "qq": true, + "weixin": true, + "line": true, + "feishu": true, + "onebot": true, +} + +// DetectVoiceCapabilities returns ASR/TTS availability for a channel, gated by +// whether providers are configured. +func DetectVoiceCapabilities(channelName string, ch Channel, asrAvailable bool, ttsAvailable bool) VoiceCapabilities { + if ch == nil { + return VoiceCapabilities{} + } + + if vcp, ok := ch.(VoiceCapabilityProvider); ok { + caps := vcp.VoiceCapabilities() + if !asrAvailable { + caps.ASR = false + } + if !ttsAvailable { + caps.TTS = false + } + return caps + } + + caps := VoiceCapabilities{} + if asrAvailable { + caps.ASR = asrCapableChannels[channelName] + } + if ttsAvailable { + if _, ok := ch.(MediaSender); ok { + caps.TTS = true + } + } + + return caps +} diff --git a/pkg/channels/weixin/weixin.go b/pkg/channels/weixin/weixin.go index fdafb02c2..937471db9 100644 --- a/pkg/channels/weixin/weixin.go +++ b/pkg/channels/weixin/weixin.go @@ -357,3 +357,8 @@ func (c *WeixinChannel) Send(ctx context.Context, msg bus.OutboundMessage) error return nil } + +// VoiceCapabilities returns the voice capabilities of the channel. +func (c *WeixinChannel) VoiceCapabilities() channels.VoiceCapabilities { + return channels.VoiceCapabilities{ASR: true, TTS: true} +} diff --git a/pkg/config/config.go b/pkg/config/config.go index 533f45a44..00e710e06 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -912,6 +912,7 @@ type ToolsConfig struct { Message ToolConfig `json:"message" yaml:"-" envPrefix:"PICOCLAW_TOOLS_MESSAGE_"` ReadFile ReadFileToolConfig `json:"read_file" yaml:"-" envPrefix:"PICOCLAW_TOOLS_READ_FILE_"` SendFile ToolConfig `json:"send_file" yaml:"-" envPrefix:"PICOCLAW_TOOLS_SEND_FILE_"` + SendTTS ToolConfig `json:"send_tts" yaml:"-" envPrefix:"PICOCLAW_TOOLS_SEND_TTS_"` Spawn ToolConfig `json:"spawn" yaml:"-" envPrefix:"PICOCLAW_TOOLS_SPAWN_"` SpawnStatus ToolConfig `json:"spawn_status" yaml:"-" envPrefix:"PICOCLAW_TOOLS_SPAWN_STATUS_"` SPI ToolConfig `json:"spi" yaml:"-" envPrefix:"PICOCLAW_TOOLS_SPI_"` @@ -1367,6 +1368,8 @@ func (t *ToolsConfig) IsToolEnabled(name string) bool { return t.WebFetch.Enabled case "send_file": return t.SendFile.Enabled + case "send_tts": + return t.SendTTS.Enabled case "write_file": return t.WriteFile.Enabled case "mcp": diff --git a/pkg/config/defaults.go b/pkg/config/defaults.go index bc4ab0649..272e825b1 100644 --- a/pkg/config/defaults.go +++ b/pkg/config/defaults.go @@ -444,6 +444,9 @@ func DefaultConfig() *Config { SendFile: ToolConfig{ Enabled: true, }, + SendTTS: ToolConfig{ + Enabled: false, + }, MCP: MCPConfig{ ToolConfig: ToolConfig{ Enabled: false, diff --git a/pkg/gateway/gateway.go b/pkg/gateway/gateway.go index 03d7dfe0c..c06ac5ac6 100644 --- a/pkg/gateway/gateway.go +++ b/pkg/gateway/gateway.go @@ -6,12 +6,15 @@ import ( "os" "os/signal" "path/filepath" + "sort" "sync" "sync/atomic" "syscall" "time" "github.com/sipeed/picoclaw/pkg/agent" + "github.com/sipeed/picoclaw/pkg/audio/asr" + "github.com/sipeed/picoclaw/pkg/audio/tts" "github.com/sipeed/picoclaw/pkg/bus" "github.com/sipeed/picoclaw/pkg/channels" _ "github.com/sipeed/picoclaw/pkg/channels/dingtalk" @@ -39,7 +42,6 @@ import ( "github.com/sipeed/picoclaw/pkg/providers" "github.com/sipeed/picoclaw/pkg/state" "github.com/sipeed/picoclaw/pkg/tools" - "github.com/sipeed/picoclaw/pkg/voice" ) const ( @@ -59,6 +61,7 @@ type services struct { ChannelManager *channels.Manager DeviceService *devices.Service HealthServer *health.Server + VoiceAgentCancel context.CancelFunc manualReloadChan chan struct{} reloading atomic.Bool } @@ -67,6 +70,27 @@ type startupBlockedProvider struct { reason string } +func logChannelVoiceCapabilities(cm *channels.Manager, asrAvailable bool, ttsAvailable bool) { + if cm == nil { + return + } + + names := cm.GetEnabledChannels() + sort.Strings(names) + for _, name := range names { + ch, ok := cm.GetChannel(name) + if !ok { + continue + } + caps := channels.DetectVoiceCapabilities(name, ch, asrAvailable, ttsAvailable) + logger.InfoCF("voice", "Channel voice capabilities", map[string]any{ + "channel": name, + "asr": caps.ASR, + "tts": caps.TTS, + }) + } +} + func (p *startupBlockedProvider) Chat( _ context.Context, _ []providers.Message, @@ -301,11 +325,14 @@ func setupAndStartServices( agentLoop.SetChannelManager(runningServices.ChannelManager) agentLoop.SetMediaStore(runningServices.MediaStore) - if transcriber := voice.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()}) } + ttsAvailable := tts.DetectTTS(cfg) != nil + enabledChannels := runningServices.ChannelManager.GetEnabledChannels() if len(enabledChannels) > 0 { fmt.Printf("✓ Channels enabled: %s\n", enabledChannels) @@ -321,6 +348,16 @@ func setupAndStartServices( return nil, fmt.Errorf("error starting channels: %w", err) } + logChannelVoiceCapabilities(runningServices.ChannelManager, transcriber != nil, ttsAvailable) + + if transcriber != nil { + // Start Voice Agent Orchestrator after channels are ready. + vaCtx, vaCancel := context.WithCancel(context.Background()) + runningServices.VoiceAgentCancel = vaCancel + voiceAgent := asr.NewAgent(msgBus, transcriber) + voiceAgent.Start(vaCtx) + } + fmt.Printf( "✓ Health endpoints available at http://%s:%d/health, /ready and /reload (POST)\n", cfg.Gateway.Host, @@ -350,6 +387,9 @@ func stopAndCleanupServices(runningServices *services, shutdownTimeout time.Dura if !isReload && runningServices.ChannelManager != nil { runningServices.ChannelManager.StopAll(shutdownCtx) } + if runningServices.VoiceAgentCancel != nil { + runningServices.VoiceAgentCancel() + } if runningServices.DeviceService != nil { runningServices.DeviceService.Stop() } @@ -527,14 +567,23 @@ func restartServices( fmt.Println(" ✓ Device event service restarted") } - transcriber := voice.DetectTranscriber(cfg) + transcriber := asr.DetectTranscriber(cfg) al.SetTranscriber(transcriber) if transcriber != nil { logger.InfoCF("voice", "Transcription re-enabled (agent-level)", map[string]any{"provider": transcriber.Name()}) + + // Start Voice Agent Orchestrator on reload + vaCtx, vaCancel := context.WithCancel(context.Background()) + runningServices.VoiceAgentCancel = vaCancel + voiceAgent := asr.NewAgent(msgBus, transcriber) + voiceAgent.Start(vaCtx) } else { logger.InfoCF("voice", "Transcription disabled", nil) } + ttsAvailable := tts.DetectTTS(cfg) != nil + logChannelVoiceCapabilities(runningServices.ChannelManager, transcriber != nil, ttsAvailable) + return nil } diff --git a/pkg/tools/tts_send.go b/pkg/tools/tts_send.go new file mode 100644 index 000000000..3d569e3f7 --- /dev/null +++ b/pkg/tools/tts_send.go @@ -0,0 +1,82 @@ +package tools + +import ( + "context" + "strings" + + "github.com/sipeed/picoclaw/pkg/audio/tts" + "github.com/sipeed/picoclaw/pkg/media" +) + +type SendTTSTool struct { + provider tts.TTSProvider + mediaStore media.MediaStore +} + +func NewSendTTSTool(provider tts.TTSProvider, store media.MediaStore) *SendTTSTool { + return &SendTTSTool{ + provider: provider, + mediaStore: store, + } +} + +func (t *SendTTSTool) Name() string { return "send_tts" } + +func (t *SendTTSTool) Description() string { + return "Synthesize speech from text and send it as an audio file to the user." +} + +func (t *SendTTSTool) Parameters() map[string]any { + return map[string]any{ + "type": "object", + "properties": map[string]any{ + "text": map[string]any{ + "type": "string", + "description": "The text to synthesize into speech. NOTE: Reply in a highly concise, conversational, oral style suitable for text-to-speech. Do not use markdown, emojis, asterisks, or code blocks. Speak naturally.", + }, + "filename": map[string]any{ + "type": "string", + "description": "Optional filename for the audio file (e.g., response.ogg).", + }, + }, + "required": []string{"text"}, + } +} + +func (t *SendTTSTool) SetMediaStore(store media.MediaStore) { + t.mediaStore = store +} + +func (t *SendTTSTool) Execute(ctx context.Context, args map[string]any) *ToolResult { + text, _ := args["text"].(string) + text = strings.TrimSpace(text) + if text == "" { + return ErrorResult("text is required") + } + + channel := ToolChannel(ctx) + chatID := ToolChatID(ctx) + filename, _ := args["filename"].(string) + + ref, err := tts.SynthesizeAndStore( + ctx, + t.provider, + t.mediaStore, + text, + filename, + channel, + chatID, + ) + if err != nil { + return ErrorResult(err.Error()).WithError(err) + } + + // Return with ForUser set to original text, Media containing the audio ref, + // and mark as ResponseHandled so the audio is sent immediately without LLM intervention. + return &ToolResult{ + ForLLM: "TTS audio sent", + ForUser: text, + Media: []string{ref}, + ResponseHandled: true, + } +} diff --git a/pkg/voice/groq_transcriber_test.go b/pkg/voice/groq_transcriber_test.go deleted file mode 100644 index fdcaa7580..000000000 --- a/pkg/voice/groq_transcriber_test.go +++ /dev/null @@ -1,84 +0,0 @@ -package voice - -import ( - "context" - "encoding/json" - "net/http" - "net/http/httptest" - "os" - "path/filepath" - "testing" -) - -var _ Transcriber = (*GroqTranscriber)(nil) - -func TestGroqTranscriberName(t *testing.T) { - tr := NewGroqTranscriber("sk-test") - if got := tr.Name(); got != "groq" { - t.Errorf("Name() = %q, want %q", got, "groq") - } -} - -func TestGroqTranscribe(t *testing.T) { - // Write a minimal fake audio file so the transcriber can open and send it. - tmpDir := t.TempDir() - audioPath := filepath.Join(tmpDir, "clip.ogg") - if err := os.WriteFile(audioPath, []byte("fake-audio-data"), 0o644); err != nil { - t.Fatalf("failed to write fake audio file: %v", err) - } - - t.Run("success", func(t *testing.T) { - srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - if r.URL.Path != "/audio/transcriptions" { - t.Errorf("unexpected path: %s", r.URL.Path) - } - if r.Header.Get("Authorization") != "Bearer sk-test" { - t.Errorf("unexpected Authorization header: %s", r.Header.Get("Authorization")) - } - w.Header().Set("Content-Type", "application/json") - _ = json.NewEncoder(w).Encode(TranscriptionResponse{ - Text: "hello world", - Language: "en", - Duration: 1.5, - }) - })) - defer srv.Close() - - tr := NewGroqTranscriber("sk-test") - tr.apiBase = srv.URL - - resp, err := tr.Transcribe(context.Background(), audioPath) - if err != nil { - t.Fatalf("Transcribe() error: %v", err) - } - if resp.Text != "hello world" { - t.Errorf("Text = %q, want %q", resp.Text, "hello world") - } - if resp.Language != "en" { - t.Errorf("Language = %q, want %q", resp.Language, "en") - } - }) - - t.Run("api error", func(t *testing.T) { - srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - http.Error(w, `{"error":"invalid_api_key"}`, http.StatusUnauthorized) - })) - defer srv.Close() - - tr := NewGroqTranscriber("sk-bad") - tr.apiBase = srv.URL - - _, err := tr.Transcribe(context.Background(), audioPath) - if err == nil { - t.Fatal("expected error for non-200 response, got nil") - } - }) - - t.Run("missing file", func(t *testing.T) { - tr := NewGroqTranscriber("sk-test") - _, err := tr.Transcribe(context.Background(), filepath.Join(tmpDir, "nonexistent.ogg")) - if err == nil { - t.Fatal("expected error for missing file, got nil") - } - }) -}