opencode/picoclaw
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor: update ASR and TTS implementations

Browse source
This commit is contained in:
Huaaudio 2026-03-28 01:25:25 +01:00
parent 60d7ec20a5
commit 01d550eb6e
39 changed files with 2104 additions and 151 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
.golangci.yaml
View file

@ -61,6 +61,9 @@ linters:
- usestdlibvars
- usetesting
settings:
gomoddirectives:
replace-allow-list:
- github.com/bwmarrin/discordgo
errcheck:
check-type-assertions: true
check-blank: true

3
config/config.example.json
View file

@ -414,6 +414,9 @@
"read_file": {
"enabled": true
},
"send_tts": {
"enabled": false
},
"spawn": {
"enabled": true
},

6
go.mod
View file

@ -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

13
go.sum
View file

@ -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=

68
pkg/agent/loop.go
View file

@ -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++ {
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

252
pkg/audio/asr/agent.go Normal file
View file

@ -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})
}
}

196
pkg/audio/asr/agent_test.go Normal file
View file

@ -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)
}

5
pkg/voice/transcriber.go → pkg/audio/asr/asr.go
View file

@ -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())
}
}

2
pkg/voice/transcriber_test.go → pkg/audio/asr/asr_test.go
View file

@ -1,4 +1,4 @@
package voice
package asr
import (
"testing"

2
pkg/voice/audio_model_transcriber.go → pkg/audio/asr/audio_model_transcriber.go
View file

@ -1,4 +1,4 @@
package voice
package asr
import (
"context"

2
pkg/voice/audio_model_transcriber_test.go → pkg/audio/asr/audio_model_transcriber_test.go
View file

@ -1,4 +1,4 @@
package voice
package asr
import (
"context"

2
pkg/voice/elevenlabs_transcriber.go → pkg/audio/asr/elevenlabs_transcriber.go
View file

@ -1,4 +1,4 @@
package voice
package asr
import (
"bytes"

2
pkg/voice/elevenlabs_transcriber_test.go → pkg/audio/asr/elevenlabs_transcriber_test.go
View file

@ -1,4 +1,4 @@
package voice
package asr
import (
"context"

97
pkg/voice/groq_transcriber.go → pkg/audio/asr/groq_transcriber.go
View file

@ -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)

57
pkg/audio/ogg.go Normal file
View file

@ -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()
}
}
}
}
}

146
pkg/audio/ogg_test.go Normal file
View file

@ -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())
}
})
}
}

96
pkg/audio/sentence.go Normal file
View file

@ -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
}

69
pkg/audio/sentence_test.go Normal file
View file

@ -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)
}
})
}
}

121
pkg/audio/tts/openai_tts.go Normal file
View file

@ -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
}

111
pkg/audio/tts/tts.go Normal file
View file

@ -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
}

129
pkg/audio/tts/tts_test.go Normal file
View file

@ -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)
}
}

28
pkg/bus/bus.go
View file

@ -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{

23
pkg/bus/types.go
View file

@ -34,6 +34,7 @@ type OutboundMessage struct {
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"
}

170
pkg/channels/discord/discord.go
View file

@ -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}
}

7
pkg/channels/discord/init.go
View file

@ -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
})
}

313
pkg/channels/discord/voice.go Normal file
View file

@ -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(),
})
}
}
}
}

7
pkg/channels/feishu/common.go
View file

@ -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}
}

5
pkg/channels/line/line.go
View file

@ -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}
}

5
pkg/channels/matrix/matrix.go
View file

@ -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}
}

5
pkg/channels/onebot/onebot.go
View file

@ -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}
}

5
pkg/channels/qq/qq.go
View file

@ -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}
}

5
pkg/channels/telegram/telegram.go
View file

@ -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}
}

58
pkg/channels/voice_capabilities.go Normal file
View file

@ -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
}

5
pkg/channels/weixin/weixin.go
View file

@ -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}
}

3
pkg/config/config.go
View file

@ -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":

3
pkg/config/defaults.go
View file

@ -444,6 +444,9 @@ func DefaultConfig() *Config {
SendFile: ToolConfig{
Enabled: true,
},
SendTTS: ToolConfig{
Enabled: false,
},
MCP: MCPConfig{
ToolConfig: ToolConfig{
Enabled: false,

55
pkg/gateway/gateway.go
View file

@ -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
}

82
pkg/tools/tts_send.go Normal file
View file

@ -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,
}
}

84
pkg/voice/groq_transcriber_test.go
View file

@ -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")
}
})
}