diff --git a/cmd/gateway_consumer_normal.go b/cmd/gateway_consumer_normal.go index 825772bb..7d08d910 100644 --- a/cmd/gateway_consumer_normal.go +++ b/cmd/gateway_consumer_normal.go @@ -12,6 +12,7 @@ import ( "github.com/nextlevelbuilder/goclaw/internal/agent" "github.com/nextlevelbuilder/goclaw/internal/bus" "github.com/nextlevelbuilder/goclaw/internal/channels" + "github.com/nextlevelbuilder/goclaw/internal/channels/telegram/voiceguard" "github.com/nextlevelbuilder/goclaw/internal/config" "github.com/nextlevelbuilder/goclaw/internal/i18n" "github.com/nextlevelbuilder/goclaw/internal/scheduler" @@ -321,7 +322,7 @@ func processNormalMessage( }) // Handle result asynchronously to not block the flush callback. - go func(agentKey, channel, chatID, session, rID string, meta map[string]string, blockReplyEnabled bool, ptd *tools.PendingTeamDispatch) { + go func(agentKey, channel, chatID, session, rID, peerKind, inboundContent string, meta map[string]string, blockReplyEnabled bool, ptd *tools.PendingTeamDispatch) { outcome := <-outCh // Release team create lock — tasks already visible in DB, other goroutines can list. @@ -396,11 +397,20 @@ func processNormalMessage( return } + // Sanitize voice agent replies: replace technical errors with user-friendly fallback. + replyContent := voiceguard.SanitizeReply( + cfg.Channels.Telegram.VoiceAgentID, agentKey, + channel, peerKind, inboundContent, outcome.Result.Content, + cfg.Channels.Telegram.AudioGuardFallbackTranscript, + cfg.Channels.Telegram.AudioGuardFallbackNoTranscript, + cfg.Channels.Telegram.AudioGuardErrorMarkers, + ) + // Publish response back to the channel outMsg := bus.OutboundMessage{ Channel: channel, ChatID: chatID, - Content: outcome.Result.Content, + Content: replyContent, Metadata: meta, } @@ -410,7 +420,7 @@ func processNormalMessage( // Auto-set followup when lead agent replies on a real channel with in_progress tasks. if teamStore != nil && channel != tools.ChannelSystem && channel != tools.ChannelTeammate && channel != tools.ChannelDashboard { - go autoSetFollowup(ctx, teamStore, agentStore, agentKey, channel, chatID, outcome.Result.Content) + go autoSetFollowup(ctx, teamStore, agentStore, agentKey, channel, chatID, replyContent) } - }(agentID, msg.Channel, msg.ChatID, sessionKey, runID, outMeta, blockReply, ptd) + }(agentID, msg.Channel, msg.ChatID, sessionKey, runID, peerKind, msg.Content, outMeta, blockReply, ptd) } diff --git a/internal/channels/media/stt_client.go b/internal/channels/media/stt_client.go index d7a689f5..fd166385 100644 --- a/internal/channels/media/stt_client.go +++ b/internal/channels/media/stt_client.go @@ -11,6 +11,7 @@ import ( "net/http" "os" "path/filepath" + "sync" "time" ) @@ -22,6 +23,26 @@ const ( sttTranscribeEndpoint = "/transcribe_audio" ) +var ( + sttClient *http.Client + sttClientOnce sync.Once + sttSem = make(chan struct{}, 4) // max 4 concurrent STT calls +) + +// getSTTClient returns a shared HTTP client with connection pooling for STT requests. +func getSTTClient() *http.Client { + sttClientOnce.Do(func() { + sttClient = &http.Client{ + Timeout: 60 * time.Second, // defensive cap in case caller context has no deadline + Transport: &http.Transport{ + MaxIdleConnsPerHost: 4, + IdleConnTimeout: 90 * time.Second, + }, + } + }) + return sttClient +} + // STTConfig holds configuration for the Speech-to-Text proxy service. type STTConfig struct { ProxyURL string // base URL of the STT proxy (e.g. "http://localhost:8080") @@ -95,8 +116,15 @@ func TranscribeAudio(ctx context.Context, cfg STTConfig, filePath string) (strin slog.Debug("stt: calling proxy", "url", url, "file", filepath.Base(filePath)) - client := &http.Client{} - resp, err := client.Do(req) + // Acquire concurrency slot (blocks if 4 calls already in-flight). + select { + case sttSem <- struct{}{}: + defer func() { <-sttSem }() + case <-reqCtx.Done(): + return "", fmt.Errorf("stt: context cancelled waiting for concurrency slot: %w", reqCtx.Err()) + } + + resp, err := getSTTClient().Do(req) if err != nil { return "", fmt.Errorf("stt: request to %q failed: %w", url, err) } diff --git a/internal/channels/telegram/voiceguard/guard.go b/internal/channels/telegram/voiceguard/guard.go new file mode 100644 index 00000000..9930b815 --- /dev/null +++ b/internal/channels/telegram/voiceguard/guard.go @@ -0,0 +1,116 @@ +// Package voiceguard intercepts technical error language in voice agent replies +// and replaces it with user-friendly fallback messages. Pure string transformation +// with zero dependencies on Telegram SDK or message bus. +package voiceguard + +import ( + "regexp" + "strings" +) + +// defaultErrorMarkers are patterns that indicate a technical/system error in the reply. +// Matched case-insensitively against the full reply text. +var defaultErrorMarkers = []string{ + // English + "system error", + "exit status", + "rate limit", + "tool error", + "service unavailable", + "technical issue", + // Vietnamese + "vấn đề kỹ thuật", + "lỗi hệ thống", + "vấn đề hệ thống", +} + +const ( + defaultFallbackTranscript = `I heard you say: "%s". Let me process that — please try again in a moment!` + defaultFallbackNoTranscript = "I received your voice message but had trouble processing it. Please try again!" +) + +// transcriptRe extracts text between ... tags. +var transcriptRe = regexp.MustCompile(`(?s)\s*(.*?)\s*`) + +// SanitizeReply checks whether a voice agent reply contains technical error language +// and replaces it with a user-friendly fallback. Returns the original reply unchanged +// when guard conditions are not met or the reply is clean. +// +// Guard conditions (all must be true): +// - voiceAgentID is non-empty and matches agentID +// - channel is "telegram" +// - peerKind is "direct" +// - inbound contains or +func SanitizeReply( + voiceAgentID, agentID, channel, peerKind, inbound, reply string, + fallbackTranscript, fallbackNoTranscript string, + errorMarkers []string, +) string { + // Guard: feature not configured or agent mismatch. + if voiceAgentID == "" || agentID != voiceAgentID { + return reply + } + // Guard: only Telegram DMs with audio/voice media. + if channel != "telegram" || peerKind != "direct" { + return reply + } + if !hasAudioTag(inbound) { + return reply + } + // Guard: reply is clean (no technical error language). + if !containsErrorLanguage(reply, errorMarkers) { + return reply + } + + // Build fallback message. + transcript := extractTranscript(inbound) + if transcript != "" { + tpl := fallbackTranscript + if tpl == "" { + tpl = defaultFallbackTranscript + } + if strings.Contains(tpl, "%s") { + return strings.Replace(tpl, "%s", transcript, 1) + } + // Template has no placeholder — return it as-is to avoid fmt garbage. + return tpl + } + + fb := fallbackNoTranscript + if fb == "" { + fb = defaultFallbackNoTranscript + } + return fb +} + +// hasAudioTag checks whether the inbound message contains a voice or audio media tag. +func hasAudioTag(inbound string) bool { + return strings.Contains(inbound, " 0 { + markers = customMarkers + } + for _, m := range markers { + if strings.Contains(lower, strings.ToLower(m)) { + return true + } + } + return false +} + +// extractTranscript extracts the transcript text from ... tags +// in the inbound message. Returns empty string if no transcript is found. +func extractTranscript(inbound string) string { + m := transcriptRe.FindStringSubmatch(inbound) + if len(m) < 2 { + return "" + } + // Collapse internal whitespace/newlines into single spaces. + return strings.Join(strings.Fields(m[1]), " ") +} diff --git a/internal/channels/telegram/voiceguard/guard_test.go b/internal/channels/telegram/voiceguard/guard_test.go new file mode 100644 index 00000000..9cb5ef04 --- /dev/null +++ b/internal/channels/telegram/voiceguard/guard_test.go @@ -0,0 +1,226 @@ +package voiceguard + +import ( + "strings" + "testing" +) + +// --------------------------------------------------------------------------- +// SanitizeReply — passthrough cases +// --------------------------------------------------------------------------- + +func TestSanitize_PassThrough_WrongAgent(t *testing.T) { + got := SanitizeReply("voice-agent", "other-agent", "telegram", "direct", + "…", "system error occurred", "", "", nil) + if got != "system error occurred" { + t.Errorf("expected passthrough, got %q", got) + } +} + +func TestSanitize_PassThrough_EmptyVoiceAgentID(t *testing.T) { + got := SanitizeReply("", "voice-agent", "telegram", "direct", + "…", "exit status 1", "", "", nil) + if got != "exit status 1" { + t.Errorf("expected passthrough when VoiceAgentID empty, got %q", got) + } +} + +func TestSanitize_PassThrough_NonTelegram(t *testing.T) { + got := SanitizeReply("voice-agent", "voice-agent", "discord", "direct", + "…", "rate limit exceeded", "", "", nil) + if got != "rate limit exceeded" { + t.Errorf("expected passthrough for non-telegram channel, got %q", got) + } +} + +func TestSanitize_PassThrough_GroupChat(t *testing.T) { + got := SanitizeReply("voice-agent", "voice-agent", "telegram", "group", + "…", "system error occurred", "", "", nil) + if got != "system error occurred" { + t.Errorf("expected passthrough for group chat, got %q", got) + } +} + +func TestSanitize_PassThrough_NoAudioTag(t *testing.T) { + got := SanitizeReply("voice-agent", "voice-agent", "telegram", "direct", + "just a regular text message", "system error occurred", "", "", nil) + if got != "system error occurred" { + t.Errorf("expected passthrough when no audio tag, got %q", got) + } +} + +func TestSanitize_PassThrough_CleanReply(t *testing.T) { + got := SanitizeReply("voice-agent", "voice-agent", "telegram", "direct", + "…", "Great job! Your pronunciation is improving.", "", "", nil) + if got != "Great job! Your pronunciation is improving." { + t.Errorf("expected clean reply passthrough, got %q", got) + } +} + +// --------------------------------------------------------------------------- +// SanitizeReply — error detection + fallback +// --------------------------------------------------------------------------- + +func TestSanitize_ErrorWithTranscript_DefaultFallback(t *testing.T) { + inbound := `I usually wake up at seven` + got := SanitizeReply("voice-agent", "voice-agent", "telegram", "direct", + inbound, "system error: tool execution failed", "", "", nil) + + if !strings.Contains(got, "I usually wake up at seven") { + t.Errorf("expected transcript in fallback, got: %q", got) + } + if strings.Contains(got, "system error") { + t.Errorf("technical error leaked into fallback: %q", got) + } +} + +func TestSanitize_ErrorWithTranscript_CustomFallback(t *testing.T) { + inbound := `hello world` + got := SanitizeReply("voice-agent", "voice-agent", "telegram", "direct", + inbound, "rate limit exceeded", + "Transcript received: %s. Please send again!", "", nil) + + want := "Transcript received: hello world. Please send again!" + if got != want { + t.Errorf("expected %q, got %q", want, got) + } +} + +func TestSanitize_ErrorWithTranscript_CustomFallbackNoPlaceholder(t *testing.T) { + customTpl := "Please resend your voice note, there was a small hiccup!" + inbound := `hello world` + got := SanitizeReply("voice-agent", "voice-agent", "telegram", "direct", + inbound, "system error: tool execution failed", customTpl, "", nil) + + if got != customTpl { + t.Errorf("expected clean fallback %q, got %q", customTpl, got) + } + if strings.Contains(got, "%!") { + t.Errorf("fmt.Sprintf garbage leaked into output: %q", got) + } +} + +func TestSanitize_ErrorNoTranscript_DefaultFallback(t *testing.T) { + got := SanitizeReply("voice-agent", "voice-agent", "telegram", "direct", + "…", "exit status 1", "", "", nil) + + if strings.Contains(got, "exit status") { + t.Errorf("technical error leaked into fallback: %q", got) + } + if got == "" { + t.Error("expected non-empty fallback, got empty string") + } +} + +func TestSanitize_ErrorNoTranscript_CustomFallback(t *testing.T) { + custom := "Sorry, please resend your voice note." + got := SanitizeReply("voice-agent", "voice-agent", "telegram", "direct", + "…", "tool error: service unavailable", "", custom, nil) + + if got != custom { + t.Errorf("expected custom no-transcript fallback %q, got %q", custom, got) + } +} + +func TestSanitize_MediaAudioTag(t *testing.T) { + inbound := `good morning` + got := SanitizeReply("voice-agent", "voice-agent", "telegram", "direct", + inbound, "rate limit: too many requests", "", "", nil) + + if strings.Contains(got, "rate limit") { + t.Errorf("technical error leaked: %q", got) + } + if !strings.Contains(got, "good morning") { + t.Errorf("expected transcript in fallback, got: %q", got) + } +} + +func TestSanitize_CustomErrorMarkers(t *testing.T) { + markers := []string{"custom failure", "oops"} + // Default marker should NOT trigger when custom markers are set. + got := SanitizeReply("voice-agent", "voice-agent", "telegram", "direct", + "…", "system error occurred", "", "", markers) + if got != "system error occurred" { + t.Errorf("expected passthrough (custom markers don't include 'system error'), got %q", got) + } + // Custom marker should trigger. + got = SanitizeReply("voice-agent", "voice-agent", "telegram", "direct", + "…", "oops something went wrong", "", "", markers) + if got == "oops something went wrong" { + t.Error("expected fallback for custom marker 'oops', got passthrough") + } +} + +// --------------------------------------------------------------------------- +// containsErrorLanguage +// --------------------------------------------------------------------------- + +func TestContainsErrorLanguage_Positives(t *testing.T) { + cases := []string{ + "vấn đề kỹ thuật xảy ra", + "lỗi hệ thống", + "vấn đề hệ thống", + "technical issue detected", + "system error: something broke", + "exit status 1", + "rate limit exceeded", + "tool error: execution failed", + "SYSTEM ERROR occurred", // mixed case + } + for _, s := range cases { + if !containsErrorLanguage(s, nil) { + t.Errorf("expected true for %q, got false", s) + } + } +} + +func TestContainsErrorLanguage_Negatives(t *testing.T) { + cases := []string{ + "", + "Great job!", + "Your pronunciation is improving.", + "Please try again.", + "I heard you say: hello world.", + } + for _, s := range cases { + if containsErrorLanguage(s, nil) { + t.Errorf("expected false for %q, got true", s) + } + } +} + +// --------------------------------------------------------------------------- +// extractTranscript +// --------------------------------------------------------------------------- + +func TestExtractTranscript_Present(t *testing.T) { + cases := []struct { + input string + want string + }{ + {`hello world`, "hello world"}, + {` spaces around `, "spaces around"}, + {"\n\nMulti\nline\ntranscript\n\n", "Multi line transcript"}, + {"only transcript", "only transcript"}, + } + for _, tc := range cases { + got := extractTranscript(tc.input) + if got != tc.want { + t.Errorf("input %q: expected %q, got %q", tc.input, tc.want, got) + } + } +} + +func TestExtractTranscript_Absent(t *testing.T) { + cases := []string{ + "…", + "plain text message", + "", + } + for _, s := range cases { + got := extractTranscript(s) + if got != "" { + t.Errorf("expected empty transcript for %q, got %q", s, got) + } + } +} diff --git a/internal/config/config_channels.go b/internal/config/config_channels.go index 0d2e1dfb..fc3f1b5f 100644 --- a/internal/config/config_channels.go +++ b/internal/config/config_channels.go @@ -54,6 +54,12 @@ type TelegramConfig struct { // agent instead of the default channel agent. Requires the named agent to exist in the config. VoiceAgentID string `json:"voice_agent_id,omitempty"` // agent ID to route voice inbound to (e.g. "speaking-agent") + // Audio guard: intercept technical errors in voice agent replies and replace with friendly fallbacks. + // Only active when VoiceAgentID is set. Custom error markers replace built-in defaults when provided. + AudioGuardFallbackTranscript string `json:"audio_guard_fallback_transcript,omitempty"` // fallback with %s for transcript (e.g. "I heard: \"%s\". Try again!") + AudioGuardFallbackNoTranscript string `json:"audio_guard_fallback_no_transcript,omitempty"` // fallback when no transcript available + AudioGuardErrorMarkers []string `json:"audio_guard_error_markers,omitempty"` // custom error detection markers (replaces defaults) + // Per-group (and per-topic) overrides. Key is chat ID string (e.g. "-100123456") or "*" for wildcard. // TS ref: channels.telegram.groups in src/config/types.telegram.ts. Groups map[string]*TelegramGroupConfig `json:"groups,omitempty"`