diff --git a/cmd/gateway.go b/cmd/gateway.go index 2b4f6fd7..b0765269 100644 --- a/cmd/gateway.go +++ b/cmd/gateway.go @@ -278,6 +278,10 @@ func runGateway() { browserMgr.SetCookieProvider(newStoreBrowserCookieProvider(pgStores.BrowserCookies)) } + if ttsTool != nil && pgStores.SystemConfigs != nil { + ttsTool.SetSystemConfigStore(pgStores.SystemConfigs) + } + // Recover from crashes: flip ghost 'summoning' rows to 'summon_failed'. // Summon goroutines don't survive process restart; stale DB rows would trap the UI. if pgStores.Agents != nil { diff --git a/cmd/gateway_agents.go b/cmd/gateway_agents.go index 13382311..60e76281 100644 --- a/cmd/gateway_agents.go +++ b/cmd/gateway_agents.go @@ -7,6 +7,7 @@ import ( "github.com/nextlevelbuilder/goclaw/internal/audio/elevenlabs" geminiaudio "github.com/nextlevelbuilder/goclaw/internal/audio/gemini" minimaxaudio "github.com/nextlevelbuilder/goclaw/internal/audio/minimax" + "github.com/nextlevelbuilder/goclaw/internal/audio/openaicompat" "github.com/nextlevelbuilder/goclaw/internal/bus" "github.com/nextlevelbuilder/goclaw/internal/config" "github.com/nextlevelbuilder/goclaw/internal/memory" @@ -275,6 +276,26 @@ func setupTTS(cfg *config.Config) *tts.Manager { })) } + // OpenAI-compatible self-hosted endpoint. api_base is the enable switch — + // unlike the vendor providers there is no API key to gate on, since these + // endpoints commonly have no auth. + if base := ttsCfg.OpenAICompat.APIBase; base != "" { + provider, err := openaicompat.NewTTSProvider(openaicompat.Config{ + APIBase: base, + APIKey: ttsCfg.OpenAICompat.APIKey, + TTSModel: ttsCfg.OpenAICompat.Model, + TTSVoice: ttsCfg.OpenAICompat.Voice, + TTSFormat: ttsCfg.OpenAICompat.Format, + TimeoutMs: ttsCfg.TimeoutMs, + }) + if err != nil { + slog.Warn("audio.tts: openai_compat not registered", "error", err) + } else { + mgr.RegisterProvider(provider) + slog.Info("audio.tts: openai_compat registered", "api_base", base) + } + } + if key := ttsCfg.ElevenLabs.APIKey; key != "" { mgr.RegisterProvider(tts.NewElevenLabsProvider(tts.ElevenLabsConfig{ APIKey: key, @@ -359,6 +380,29 @@ func setupAudioExtras(cfg *config.Config, mgr *tts.Manager) { } } + // STT chain. Built from what actually registered, so Transcribe never walks + // a name it will only skip with a warning. "proxy" is always appended: it is + // registered later, per channel, by BridgeLegacySTT. + var sttChain []string + + // OpenAI-compatible self-hosted endpoint, first when present: it is an + // explicit operator choice, and it keeps audio on the local network. + if base := cfg.Tts.OpenAICompat.APIBase; base != "" { + provider, err := openaicompat.NewSTTProvider(openaicompat.Config{ + APIBase: base, + APIKey: cfg.Tts.OpenAICompat.APIKey, + STTModel: cfg.Tts.OpenAICompat.STTModel, + TimeoutMs: cfg.Tts.TimeoutMs, + }) + if err != nil { + slog.Warn("audio.stt: openai_compat not registered", "error", err) + } else { + mgr.RegisterSTT(provider) + sttChain = append(sttChain, provider.Name()) + slog.Info("audio.stt: openai_compat registered", "api_base", base) + } + } + // ElevenLabs STT (Scribe v2) — reuse TTS credentials. Registered as tenant-scope // default; per-request tenant override lands via builtin_tools[stt] in Phase 5 // channel migration. Legacy per-channel STTProxyURL is bridged separately. @@ -367,7 +411,13 @@ func setupAudioExtras(cfg *config.Config, mgr *tts.Manager) { APIKey: ellKey, BaseURL: ellBase, })) - mgr.SetSTTChain([]string{"elevenlabs", "proxy"}) + sttChain = append(sttChain, "elevenlabs") slog.Info("audio.stt: elevenlabs registered") } + + if len(sttChain) > 0 { + sttChain = append(sttChain, "proxy") + mgr.SetSTTChain(sttChain) + slog.Info("audio.stt: chain configured", "chain", sttChain) + } } diff --git a/internal/audio/openaicompat/config.go b/internal/audio/openaicompat/config.go new file mode 100644 index 00000000..4dcb3a3e --- /dev/null +++ b/internal/audio/openaicompat/config.go @@ -0,0 +1,74 @@ +// Package openaicompat implements the OpenAI audio API (speech + transcriptions) +// against any OpenAI-compatible endpoint — gpu-manager, Speaches, vLLM, LocalAI, +// llama.cpp, Ollama. +// +// It is deliberately distinct from the openai package. Both speak the same wire +// format, but "openai" means api.openai.com semantics: a fixed voice set +// (alloy, echo, ...), an mp3 default, and a base URL that falls back to the +// public API. A self-hosted engine shares none of that — its voices are its own +// (e.g. Piper's "fr_FR-gilles-low"), its formats are whatever it built in, and +// there is no sane default host. +// +// Keeping them separate matters beyond tidiness: audio.IsVoiceCompatible applies +// OpenAI's voice allowlist to any provider named "openai" and silently +// substitutes the default voice for anything outside it. Routing a self-hosted +// engine through that name turns a valid local voice ID into "alloy" with no +// error. Providers without validation rules pass through untouched, which is the +// correct behaviour here. +package openaicompat + +import "errors" + +// ErrAPIBaseRequired is returned by the constructors when APIBase is empty. +// Unlike the openai package there is no public endpoint to fall back to, and +// defaulting to api.openai.com would send self-hosted traffic to a vendor. +var ErrAPIBaseRequired = errors.New("openai_compat: api_base is required") + +// providerName is the stable identifier reported to the audio Manager and used +// in chain configuration. It must not be "openai" — see the package doc. +const providerName = "openai_compat" + +// Default tuning. Formats intentionally follow the OpenAI wire defaults; an +// endpoint that cannot encode mp3 is configured with Format: "wav". +const ( + defaultTTSFormat = "mp3" + defaultTimeoutMs = 30000 +) + +// Config bundles the endpoint, credentials and defaults for both the TTS and +// STT providers. APIBase is required; everything else has a default or is +// resolved per-request. +type Config struct { + // APIBase is the OpenAI-compatible root, including any version prefix — + // e.g. "http://gpu-manager:8080/v1". Required. + APIBase string + // APIKey is sent as a Bearer token when non-empty. Self-hosted endpoints + // frequently have no auth, so an empty key omits the header entirely + // rather than sending "Bearer ". + APIKey string + // TTSModel is the default model for synthesis. Optional: some engines + // resolve the model from the requested voice instead. + TTSModel string + // TTSVoice is the default voice for synthesis. Optional. + TTSVoice string + // TTSFormat is the default response_format for synthesis. Defaults to + // "mp3" to match OpenAI. Set to "wav" or "pcm" for engines without an + // encoder. + TTSFormat string + // STTModel is the default model for transcription. Optional. + STTModel string + // TimeoutMs is the per-request timeout. Defaults to 30000. + TimeoutMs int +} + +// withDefaults returns a copy with zero-valued optional fields filled in. +// APIBase is not defaulted — its absence is an error, not a fallback. +func (c Config) withDefaults() Config { + if c.TTSFormat == "" { + c.TTSFormat = defaultTTSFormat + } + if c.TimeoutMs <= 0 { + c.TimeoutMs = defaultTimeoutMs + } + return c +} diff --git a/internal/audio/openaicompat/stt.go b/internal/audio/openaicompat/stt.go new file mode 100644 index 00000000..7aa04854 --- /dev/null +++ b/internal/audio/openaicompat/stt.go @@ -0,0 +1,211 @@ +package openaicompat + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "mime/multipart" + "net/http" + "os" + "path/filepath" + "strings" + "time" + + "github.com/nextlevelbuilder/goclaw/internal/audio" +) + +const ( + // sttMaxBytes matches the OpenAI upload limit, which compatible engines + // generally adopt. Enforced before the request so an oversized file fails + // locally instead of after a full upload. + sttMaxBytes = 25 << 20 // 25 MB + // sttDefaultTimeout is the floor for transcription, which is slower than + // synthesis and routinely exceeds the shared 30 s default. + sttDefaultTimeout = 120 * time.Second +) + +// STTProvider transcribes audio via an OpenAI-compatible +// POST /audio/transcriptions. +type STTProvider struct { + cfg Config +} + +// NewSTTProvider returns an STT provider for the configured endpoint. +// Returns ErrAPIBaseRequired when cfg.APIBase is empty. +func NewSTTProvider(cfg Config) (*STTProvider, error) { + if strings.TrimSpace(cfg.APIBase) == "" { + return nil, ErrAPIBaseRequired + } + return &STTProvider{cfg: cfg.withDefaults()}, nil +} + +// Name returns the stable provider identifier. +func (p *STTProvider) Name() string { return providerName } + +// Transcribe converts audio to text. +// +// The multipart filename is significant, not cosmetic: several compatible +// engines type the audio by its extension rather than by Content-Type, and +// reject an upload whose name carries no format they recognise. The filename is +// therefore derived from STTInput.Filename, then the base of FilePath, then the +// MIME type — never left blank. +func (p *STTProvider) Transcribe(ctx context.Context, in audio.STTInput, opts audio.STTOptions) (*audio.TranscriptResult, error) { + body, contentType, err := p.buildForm(in, opts) + if err != nil { + return nil, err + } + + url := strings.TrimRight(p.cfg.APIBase, "/") + "/audio/transcriptions" + req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, body) + if err != nil { + return nil, fmt.Errorf("openai_compat stt: create request: %w", err) + } + req.Header.Set("Content-Type", contentType) + if p.cfg.APIKey != "" { + req.Header.Set("Authorization", "Bearer "+p.cfg.APIKey) + } + + timeout := sttDefaultTimeout + if opts.TimeoutMs > 0 { + timeout = time.Duration(opts.TimeoutMs) * time.Millisecond + } + hc := &http.Client{Timeout: timeout} + resp, err := hc.Do(req) + if err != nil { + return nil, fmt.Errorf("openai_compat stt: http request: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + errBody, _ := io.ReadAll(io.LimitReader(resp.Body, errBodyLimit)) + return nil, fmt.Errorf("openai_compat stt: endpoint error %d: %s", resp.StatusCode, strings.TrimSpace(string(errBody))) + } + + // verbose_json is a superset of json; language and duration are absent from + // a plain json response and stay zero-valued, which the Manager tolerates. + var result struct { + Text string `json:"text"` + Language string `json:"language"` + Duration float64 `json:"duration"` + } + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return nil, fmt.Errorf("openai_compat stt: parse response: %w", err) + } + + language := result.Language + if language == "" { + language = opts.Language + } + return &audio.TranscriptResult{ + Text: result.Text, + Language: language, + Duration: result.Duration, + Provider: providerName, + }, nil +} + +// buildForm assembles the multipart body, reading the audio from FilePath or +// Bytes. Returns the body and its Content-Type header value. +func (p *STTProvider) buildForm(in audio.STTInput, opts audio.STTOptions) (io.Reader, string, error) { + var buf bytes.Buffer + mw := multipart.NewWriter(&buf) + + if model := firstNonEmpty(opts.ModelID, p.cfg.STTModel); model != "" { + if err := mw.WriteField("model", model); err != nil { + return nil, "", fmt.Errorf("openai_compat stt: write model field: %w", err) + } + } + if opts.Language != "" { + if err := mw.WriteField("language", opts.Language); err != nil { + return nil, "", fmt.Errorf("openai_compat stt: write language field: %w", err) + } + } + if err := mw.WriteField("response_format", "json"); err != nil { + return nil, "", fmt.Errorf("openai_compat stt: write response_format field: %w", err) + } + + fw, err := mw.CreateFormFile("file", sttFilename(in)) + if err != nil { + return nil, "", fmt.Errorf("openai_compat stt: create form file: %w", err) + } + if err := writeAudio(fw, in); err != nil { + return nil, "", err + } + if err := mw.Close(); err != nil { + return nil, "", fmt.Errorf("openai_compat stt: close multipart writer: %w", err) + } + return &buf, mw.FormDataContentType(), nil +} + +// writeAudio copies the input audio into dst, enforcing the size cap. +func writeAudio(dst io.Writer, in audio.STTInput) error { + if in.FilePath != "" { + info, err := os.Stat(in.FilePath) + if err != nil { + return fmt.Errorf("openai_compat stt: stat file: %w", err) + } + if info.Size() > sttMaxBytes { + return fmt.Errorf("openai_compat stt: file too large (%d bytes, max %d)", info.Size(), sttMaxBytes) + } + f, err := os.Open(in.FilePath) + if err != nil { + return fmt.Errorf("openai_compat stt: open file: %w", err) + } + defer f.Close() + if _, err := io.Copy(dst, f); err != nil { + return fmt.Errorf("openai_compat stt: read file: %w", err) + } + return nil + } + + if len(in.Bytes) == 0 { + return fmt.Errorf("openai_compat stt: neither FilePath nor Bytes provided") + } + if len(in.Bytes) > sttMaxBytes { + return fmt.Errorf("openai_compat stt: audio too large (%d bytes, max %d)", len(in.Bytes), sttMaxBytes) + } + if _, err := dst.Write(in.Bytes); err != nil { + return fmt.Errorf("openai_compat stt: write audio bytes: %w", err) + } + return nil +} + +// sttFilename derives the multipart filename. See Transcribe for why a blank +// name is not an option. +func sttFilename(in audio.STTInput) string { + if in.Filename != "" { + return in.Filename + } + if in.FilePath != "" { + return filepath.Base(in.FilePath) + } + return "audio" + extFromMime(in.MimeType) +} + +// extFromMime maps a MIME type to a file extension, including the leading dot. +func extFromMime(mime string) string { + // Strip codec parameters, e.g. "audio/ogg; codecs=opus". + if idx := strings.IndexByte(mime, ';'); idx >= 0 { + mime = mime[:idx] + } + switch strings.TrimSpace(mime) { + case "audio/ogg", "audio/opus": + return ".ogg" + case "audio/mpeg", "audio/mp3": + return ".mp3" + case "audio/wav", "audio/wave", "audio/x-wav": + return ".wav" + case "audio/pcm", "audio/l16": + return ".pcm" + case "audio/mp4", "audio/m4a", "audio/x-m4a": + return ".m4a" + case "audio/webm": + return ".webm" + case "audio/flac", "audio/x-flac": + return ".flac" + default: + return ".bin" + } +} diff --git a/internal/audio/openaicompat/stt_test.go b/internal/audio/openaicompat/stt_test.go new file mode 100644 index 00000000..7654caea --- /dev/null +++ b/internal/audio/openaicompat/stt_test.go @@ -0,0 +1,267 @@ +package openaicompat_test + +import ( + "context" + "io" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/nextlevelbuilder/goclaw/internal/audio" + "github.com/nextlevelbuilder/goclaw/internal/audio/openaicompat" +) + +// sttForm captures what the fake endpoint received. +type sttForm struct { + path string + auth string + filename string + fileData string + model string + language string + format string +} + +// newSTTServer returns a fake OpenAI-compatible transcription endpoint that +// records the multipart form and replies with body. +func newSTTServer(t *testing.T, got *sttForm, status int, body string) *httptest.Server { + t.Helper() + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + got.path = r.URL.Path + got.auth = r.Header.Get("Authorization") + require.NoError(t, r.ParseMultipartForm(32<<20)) + got.model = r.FormValue("model") + got.language = r.FormValue("language") + got.format = r.FormValue("response_format") + + file, header, err := r.FormFile("file") + if err == nil { + defer file.Close() + got.filename = header.Filename + data, err := io.ReadAll(file) + require.NoError(t, err) + got.fileData = string(data) + } + + w.WriteHeader(status) + _, _ = io.WriteString(w, body) + })) + t.Cleanup(func() { srv.Close() }) + return srv +} + +func TestNewSTTProviderRequiresAPIBase(t *testing.T) { + _, err := openaicompat.NewSTTProvider(openaicompat.Config{}) + require.ErrorIs(t, err, openaicompat.ErrAPIBaseRequired) +} + +func TestSTTProviderName(t *testing.T) { + provider, err := openaicompat.NewSTTProvider(openaicompat.Config{APIBase: "http://x/v1"}) + require.NoError(t, err) + assert.Equal(t, "openai_compat", provider.Name()) +} + +func TestSTTTranscribeFromBytes(t *testing.T) { + var got sttForm + srv := newSTTServer(t, &got, http.StatusOK, `{"text":"bonjour le monde","language":"fr","duration":1.5}`) + + provider, err := openaicompat.NewSTTProvider(openaicompat.Config{ + APIBase: srv.URL + "/v1", + APIKey: "secret", + STTModel: "whisper-1", + }) + require.NoError(t, err) + + result, err := provider.Transcribe(context.Background(), + audio.STTInput{Bytes: []byte("RIFFfake"), MimeType: "audio/wav", Filename: "utterance.wav"}, + audio.STTOptions{Language: "fr"}, + ) + require.NoError(t, err) + + assert.Equal(t, "/v1/audio/transcriptions", got.path) + assert.Equal(t, "Bearer secret", got.auth) + assert.Equal(t, "utterance.wav", got.filename) + assert.Equal(t, "RIFFfake", got.fileData) + assert.Equal(t, "whisper-1", got.model) + assert.Equal(t, "fr", got.language) + assert.Equal(t, "json", got.format) + + assert.Equal(t, "bonjour le monde", result.Text) + assert.Equal(t, "fr", result.Language) + assert.InEpsilon(t, 1.5, result.Duration, 0.0001) + assert.Equal(t, "openai_compat", result.Provider) +} + +func TestSTTTranscribeFromFilePath(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "voice-note.ogg") + require.NoError(t, os.WriteFile(path, []byte("OggS-fake"), 0o600)) + + var got sttForm + srv := newSTTServer(t, &got, http.StatusOK, `{"text":"salut"}`) + + provider, err := openaicompat.NewSTTProvider(openaicompat.Config{APIBase: srv.URL}) + require.NoError(t, err) + + result, err := provider.Transcribe(context.Background(), + audio.STTInput{FilePath: path, MimeType: "audio/ogg"}, + audio.STTOptions{}, + ) + require.NoError(t, err) + + assert.Equal(t, "voice-note.ogg", got.filename, "filename falls back to the base of FilePath") + assert.Equal(t, "OggS-fake", got.fileData) + assert.Equal(t, "salut", result.Text) +} + +// TestSTTFilenameIsNeverBlank guards the extension-typing requirement: some +// compatible engines (gpu-manager among them) infer the audio format from the +// filename and reject an upload they cannot type. +func TestSTTFilenameIsNeverBlank(t *testing.T) { + tests := []struct { + name string + in audio.STTInput + wantName string + }{ + { + name: "explicit filename wins", + in: audio.STTInput{Bytes: []byte("x"), Filename: "utterance.wav", MimeType: "audio/ogg"}, + wantName: "utterance.wav", + }, + { + name: "derived from wav mime", + in: audio.STTInput{Bytes: []byte("x"), MimeType: "audio/wav"}, + wantName: "audio.wav", + }, + { + name: "derived from ogg mime", + in: audio.STTInput{Bytes: []byte("x"), MimeType: "audio/ogg"}, + wantName: "audio.ogg", + }, + { + name: "codec parameters stripped", + in: audio.STTInput{Bytes: []byte("x"), MimeType: "audio/ogg; codecs=opus"}, + wantName: "audio.ogg", + }, + { + name: "mpeg mime", + in: audio.STTInput{Bytes: []byte("x"), MimeType: "audio/mpeg"}, + wantName: "audio.mp3", + }, + { + name: "unknown mime still carries an extension", + in: audio.STTInput{Bytes: []byte("x"), MimeType: "audio/weird"}, + wantName: "audio.bin", + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + var got sttForm + srv := newSTTServer(t, &got, http.StatusOK, `{"text":"ok"}`) + + provider, err := openaicompat.NewSTTProvider(openaicompat.Config{APIBase: srv.URL}) + require.NoError(t, err) + + _, err = provider.Transcribe(context.Background(), tc.in, audio.STTOptions{}) + require.NoError(t, err) + assert.Equal(t, tc.wantName, got.filename) + }) + } +} + +func TestSTTOmitsModelWhenUnset(t *testing.T) { + var got sttForm + srv := newSTTServer(t, &got, http.StatusOK, `{"text":"ok"}`) + + provider, err := openaicompat.NewSTTProvider(openaicompat.Config{APIBase: srv.URL}) + require.NoError(t, err) + + _, err = provider.Transcribe(context.Background(), + audio.STTInput{Bytes: []byte("x"), MimeType: "audio/wav"}, audio.STTOptions{}) + require.NoError(t, err) + assert.Empty(t, got.model, "an unset model must be omitted so the endpoint picks its own") +} + +func TestSTTOptionsModelOverridesConfig(t *testing.T) { + var got sttForm + srv := newSTTServer(t, &got, http.StatusOK, `{"text":"ok"}`) + + provider, err := openaicompat.NewSTTProvider(openaicompat.Config{APIBase: srv.URL, STTModel: "config-model"}) + require.NoError(t, err) + + _, err = provider.Transcribe(context.Background(), + audio.STTInput{Bytes: []byte("x"), MimeType: "audio/wav"}, + audio.STTOptions{ModelID: "opts-model"}) + require.NoError(t, err) + assert.Equal(t, "opts-model", got.model) +} + +// TestSTTLanguageFallsBackToHint covers a plain (non-verbose) json response, +// which carries no language field. +func TestSTTLanguageFallsBackToHint(t *testing.T) { + var got sttForm + srv := newSTTServer(t, &got, http.StatusOK, `{"text":"bonjour"}`) + + provider, err := openaicompat.NewSTTProvider(openaicompat.Config{APIBase: srv.URL}) + require.NoError(t, err) + + result, err := provider.Transcribe(context.Background(), + audio.STTInput{Bytes: []byte("x"), MimeType: "audio/wav"}, + audio.STTOptions{Language: "fr"}) + require.NoError(t, err) + assert.Equal(t, "fr", result.Language) +} + +func TestSTTErrorStatus(t *testing.T) { + var got sttForm + srv := newSTTServer(t, &got, http.StatusUnsupportedMediaType, `{"error":"unsupported audio format"}`) + + provider, err := openaicompat.NewSTTProvider(openaicompat.Config{APIBase: srv.URL}) + require.NoError(t, err) + + _, err = provider.Transcribe(context.Background(), + audio.STTInput{Bytes: []byte("x"), MimeType: "audio/ogg"}, audio.STTOptions{}) + require.Error(t, err) + assert.Contains(t, err.Error(), "415") + assert.Contains(t, err.Error(), "unsupported audio format") +} + +func TestSTTRejectsEmptyInput(t *testing.T) { + provider, err := openaicompat.NewSTTProvider(openaicompat.Config{APIBase: "http://x/v1"}) + require.NoError(t, err) + + _, err = provider.Transcribe(context.Background(), audio.STTInput{}, audio.STTOptions{}) + require.Error(t, err) + assert.Contains(t, err.Error(), "neither FilePath nor Bytes") +} + +func TestSTTRejectsOversizedBytes(t *testing.T) { + provider, err := openaicompat.NewSTTProvider(openaicompat.Config{APIBase: "http://x/v1"}) + require.NoError(t, err) + + _, err = provider.Transcribe(context.Background(), + audio.STTInput{Bytes: make([]byte, (25<<20)+1), MimeType: "audio/wav"}, + audio.STTOptions{}) + require.Error(t, err) + assert.Contains(t, err.Error(), "too large") +} + +func TestSTTRejectsOversizedFile(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "big.wav") + require.NoError(t, os.WriteFile(path, make([]byte, (25<<20)+1), 0o600)) + + provider, err := openaicompat.NewSTTProvider(openaicompat.Config{APIBase: "http://x/v1"}) + require.NoError(t, err) + + _, err = provider.Transcribe(context.Background(), + audio.STTInput{FilePath: path, MimeType: "audio/wav"}, audio.STTOptions{}) + require.Error(t, err) + assert.Contains(t, err.Error(), "too large") +} diff --git a/internal/audio/openaicompat/tts.go b/internal/audio/openaicompat/tts.go new file mode 100644 index 00000000..7c583c50 --- /dev/null +++ b/internal/audio/openaicompat/tts.go @@ -0,0 +1,170 @@ +package openaicompat + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "strings" + "time" + + "github.com/nextlevelbuilder/goclaw/internal/audio" +) + +// errBodyLimit caps how much of an error response body is read into a message. +const errBodyLimit = 512 + +// TTSProvider synthesizes speech via an OpenAI-compatible POST /audio/speech. +type TTSProvider struct { + cfg Config +} + +// NewTTSProvider returns a TTS provider for the configured endpoint. +// Returns ErrAPIBaseRequired when cfg.APIBase is empty. +func NewTTSProvider(cfg Config) (*TTSProvider, error) { + if strings.TrimSpace(cfg.APIBase) == "" { + return nil, ErrAPIBaseRequired + } + return &TTSProvider{cfg: cfg.withDefaults()}, nil +} + +// Name returns the stable provider identifier. +func (p *TTSProvider) Name() string { return providerName } + +// Synthesize calls POST {api_base}/audio/speech and returns the audio bytes. +// +// Resolution order for voice, model and format is: per-request opts, then +// opts.Params, then the provider config. Voice and model are omitted from the +// request when they resolve empty — some engines pick the model from the voice, +// and sending an empty string would override that choice rather than defer to it. +func (p *TTSProvider) Synthesize(ctx context.Context, text string, opts audio.TTSOptions) (*audio.SynthResult, error) { + format := firstNonEmpty(opts.Format, paramString(opts.Params, "response_format"), p.cfg.TTSFormat) + voice := firstNonEmpty(opts.Voice, p.cfg.TTSVoice) + model := firstNonEmpty(opts.Model, p.cfg.TTSModel) + + body := map[string]any{ + "input": text, + "response_format": format, + } + if voice != "" { + body["voice"] = voice + } + if model != "" { + body["model"] = model + } + if speed := paramFloat(opts.Params, "speed"); speed > 0 { + body["speed"] = speed + } + + bodyJSON, err := json.Marshal(body) + if err != nil { + return nil, fmt.Errorf("openai_compat tts: marshal request: %w", err) + } + + url := strings.TrimRight(p.cfg.APIBase, "/") + "/audio/speech" + req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(bodyJSON)) + if err != nil { + return nil, fmt.Errorf("openai_compat tts: create request: %w", err) + } + req.Header.Set("Content-Type", "application/json") + if p.cfg.APIKey != "" { + req.Header.Set("Authorization", "Bearer "+p.cfg.APIKey) + } + + hc := &http.Client{Timeout: time.Duration(p.cfg.TimeoutMs) * time.Millisecond} + resp, err := hc.Do(req) + if err != nil { + return nil, fmt.Errorf("openai_compat tts: http request: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + errBody, _ := io.ReadAll(io.LimitReader(resp.Body, errBodyLimit)) + return nil, fmt.Errorf("openai_compat tts: endpoint error %d: %s", resp.StatusCode, strings.TrimSpace(string(errBody))) + } + + audioBytes, err := io.ReadAll(resp.Body) + if err != nil { + return nil, fmt.Errorf("openai_compat tts: read response: %w", err) + } + + ext, mime := formatToExtMime(format) + return &audio.SynthResult{ + Audio: audioBytes, + Extension: ext, + MimeType: mime, + }, nil +} + +// formatToExtMime maps an OpenAI response_format to a file extension and MIME +// type. Unknown formats are passed through as their own extension with a +// generic MIME type rather than guessed at. +func formatToExtMime(format string) (ext, mime string) { + switch format { + case "mp3": + return "mp3", "audio/mpeg" + case "opus": + return "ogg", "audio/ogg" + case "aac": + return "aac", "audio/aac" + case "flac": + return "flac", "audio/flac" + case "wav": + return "wav", "audio/wav" + case "pcm": + return "pcm", "audio/pcm" + default: + return format, "application/octet-stream" + } +} + +// firstNonEmpty returns the first non-empty string, or "" if all are empty. +func firstNonEmpty(values ...string) string { + for _, v := range values { + if v != "" { + return v + } + } + return "" +} + +// paramString reads a string param, returning "" when absent or not a string. +func paramString(params map[string]any, key string) string { + if params == nil { + return "" + } + v, ok := audio.GetNested(params, key) + if !ok { + return "" + } + s, ok := v.(string) + if !ok { + return "" + } + return s +} + +// paramFloat reads a numeric param, returning 0 when absent or not numeric. +func paramFloat(params map[string]any, key string) float64 { + if params == nil { + return 0 + } + v, ok := audio.GetNested(params, key) + if !ok { + return 0 + } + switch n := v.(type) { + case float64: + return n + case float32: + return float64(n) + case int: + return float64(n) + case int64: + return float64(n) + default: + return 0 + } +} diff --git a/internal/audio/openaicompat/tts_test.go b/internal/audio/openaicompat/tts_test.go new file mode 100644 index 00000000..3cc767f1 --- /dev/null +++ b/internal/audio/openaicompat/tts_test.go @@ -0,0 +1,209 @@ +package openaicompat_test + +import ( + "context" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/nextlevelbuilder/goclaw/internal/audio" + "github.com/nextlevelbuilder/goclaw/internal/audio/openaicompat" +) + +// piperVoice is a real Piper voice ID. It is deliberately used throughout: +// it is exactly the kind of self-hosted voice that the "openai" provider name +// would silently rewrite to "alloy". +const piperVoice = "fr_FR-gilles-low" + +func TestNewTTSProviderRequiresAPIBase(t *testing.T) { + _, err := openaicompat.NewTTSProvider(openaicompat.Config{}) + require.ErrorIs(t, err, openaicompat.ErrAPIBaseRequired) + + _, err = openaicompat.NewTTSProvider(openaicompat.Config{APIBase: " "}) + require.ErrorIs(t, err, openaicompat.ErrAPIBaseRequired, "whitespace-only api_base must not pass") +} + +func TestTTSProviderName(t *testing.T) { + provider, err := openaicompat.NewTTSProvider(openaicompat.Config{APIBase: "http://x/v1"}) + require.NoError(t, err) + assert.Equal(t, "openai_compat", provider.Name()) +} + +// TestTTSProviderNameEscapesOpenAIVoiceValidation is the regression guard for +// the reason this package exists. Named "openai", a self-hosted voice ID is +// replaced by "alloy" before the request is ever sent. +func TestTTSProviderNameEscapesOpenAIVoiceValidation(t *testing.T) { + provider, err := openaicompat.NewTTSProvider(openaicompat.Config{APIBase: "http://x/v1"}) + require.NoError(t, err) + + assert.True(t, audio.IsVoiceCompatible(provider.Name(), piperVoice), + "a self-hosted voice must survive validation under this provider name") + + filtered, changed := audio.FilterVoiceForProvider(provider.Name(), piperVoice, false) + assert.Equal(t, piperVoice, filtered) + assert.False(t, changed) + + // Contrast: the same voice under the "openai" name is silently replaced. + clobbered, changed := audio.FilterVoiceForProvider("openai", piperVoice, false) + assert.Equal(t, "alloy", clobbered) + assert.True(t, changed) +} + +func TestTTSSynthesize(t *testing.T) { + var gotBody map[string]any + var gotAuth, gotPath string + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotPath = r.URL.Path + gotAuth = r.Header.Get("Authorization") + require.NoError(t, json.NewDecoder(r.Body).Decode(&gotBody)) + w.Header().Set("Content-Type", "audio/wav") + _, _ = w.Write([]byte("RIFFfake")) + })) + t.Cleanup(func() { srv.Close() }) + + provider, err := openaicompat.NewTTSProvider(openaicompat.Config{ + APIBase: srv.URL + "/v1", + APIKey: "secret", + TTSFormat: "wav", + TTSVoice: piperVoice, + }) + require.NoError(t, err) + + result, err := provider.Synthesize(context.Background(), "bonjour", audio.TTSOptions{}) + require.NoError(t, err) + + assert.Equal(t, "/v1/audio/speech", gotPath) + assert.Equal(t, "Bearer secret", gotAuth) + assert.Equal(t, "bonjour", gotBody["input"]) + assert.Equal(t, piperVoice, gotBody["voice"], "configured voice must reach the endpoint verbatim") + assert.Equal(t, "wav", gotBody["response_format"]) + assert.NotContains(t, gotBody, "model", "an empty model must be omitted, not sent blank") + + assert.Equal(t, []byte("RIFFfake"), result.Audio) + assert.Equal(t, "wav", result.Extension) + assert.Equal(t, "audio/wav", result.MimeType) +} + +func TestTTSSynthesizeResolutionOrder(t *testing.T) { + var gotBody map[string]any + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + require.NoError(t, json.NewDecoder(r.Body).Decode(&gotBody)) + _, _ = w.Write([]byte("x")) + })) + t.Cleanup(func() { srv.Close() }) + + provider, err := openaicompat.NewTTSProvider(openaicompat.Config{ + APIBase: srv.URL, + TTSVoice: "config-voice", + TTSModel: "config-model", + TTSFormat: "wav", + }) + require.NoError(t, err) + + _, err = provider.Synthesize(context.Background(), "hi", audio.TTSOptions{ + Voice: "opts-voice", + Model: "opts-model", + Format: "pcm", + Params: map[string]any{"speed": 1.5}, + }) + require.NoError(t, err) + + assert.Equal(t, "opts-voice", gotBody["voice"], "per-request opts win over config") + assert.Equal(t, "opts-model", gotBody["model"]) + assert.Equal(t, "pcm", gotBody["response_format"]) + assert.InEpsilon(t, 1.5, gotBody["speed"], 0.0001) +} + +func TestTTSSynthesizeParamsFormatOverridesConfig(t *testing.T) { + var gotBody map[string]any + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + require.NoError(t, json.NewDecoder(r.Body).Decode(&gotBody)) + _, _ = w.Write([]byte("x")) + })) + t.Cleanup(func() { srv.Close() }) + + provider, err := openaicompat.NewTTSProvider(openaicompat.Config{ + APIBase: srv.URL, + TTSFormat: "wav", + }) + require.NoError(t, err) + + _, err = provider.Synthesize(context.Background(), "hi", audio.TTSOptions{ + Params: map[string]any{"response_format": "mp3"}, + }) + require.NoError(t, err) + assert.Equal(t, "mp3", gotBody["response_format"]) +} + +func TestTTSSynthesizeOmitsAuthWhenNoKey(t *testing.T) { + var hasAuth bool + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, hasAuth = r.Header["Authorization"] + _, _ = w.Write([]byte("x")) + })) + t.Cleanup(func() { srv.Close() }) + + provider, err := openaicompat.NewTTSProvider(openaicompat.Config{APIBase: srv.URL}) + require.NoError(t, err) + + _, err = provider.Synthesize(context.Background(), "hi", audio.TTSOptions{}) + require.NoError(t, err) + assert.False(t, hasAuth, "self-hosted endpoints without auth must not receive an empty Bearer header") +} + +func TestTTSSynthesizeErrorStatus(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusBadRequest) + _, _ = io.WriteString(w, `{"error":"response_format must be wav or pcm"}`) + })) + t.Cleanup(func() { srv.Close() }) + + provider, err := openaicompat.NewTTSProvider(openaicompat.Config{APIBase: srv.URL}) + require.NoError(t, err) + + _, err = provider.Synthesize(context.Background(), "hi", audio.TTSOptions{}) + require.Error(t, err) + assert.Contains(t, err.Error(), "400") + assert.Contains(t, err.Error(), "response_format must be wav or pcm", + "the endpoint's own message must survive so a format mismatch is diagnosable") +} + +func TestTTSFormatExtensionMapping(t *testing.T) { + tests := []struct { + format string + wantExt string + wantMIME string + }{ + {"mp3", "mp3", "audio/mpeg"}, + {"opus", "ogg", "audio/ogg"}, + {"wav", "wav", "audio/wav"}, + {"pcm", "pcm", "audio/pcm"}, + {"flac", "flac", "audio/flac"}, + {"aac", "aac", "audio/aac"}, + } + for _, tc := range tests { + t.Run(tc.format, func(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + _, _ = w.Write([]byte("x")) + })) + t.Cleanup(func() { srv.Close() }) + + provider, err := openaicompat.NewTTSProvider(openaicompat.Config{ + APIBase: srv.URL, + TTSFormat: tc.format, + }) + require.NoError(t, err) + + result, err := provider.Synthesize(context.Background(), "hi", audio.TTSOptions{}) + require.NoError(t, err) + assert.Equal(t, tc.wantExt, result.Extension) + assert.Equal(t, tc.wantMIME, result.MimeType) + }) + } +} diff --git a/internal/channels/bitrix24/channel.go b/internal/channels/bitrix24/channel.go index 69cbd089..252fdee2 100644 --- a/internal/channels/bitrix24/channel.go +++ b/internal/channels/bitrix24/channel.go @@ -56,6 +56,12 @@ type Channel struct { mentionMu sync.Mutex mentionRe *mentionMatcher + // reactions tracks the current status-reaction emoji per user message so + // OnReactionEvent can replace the previous one as the agent's status + // progresses (thinking → tool → done). Keyed by ":"; + // values are *reactionState. See reactions.go. + reactions sync.Map + // MCP lazy provisioner (Phase C). All fields nil / zero when // provisioning is disabled — channel then works exactly as before // (messages flow through without trying to mint MCP credentials). diff --git a/internal/channels/bitrix24/handle.go b/internal/channels/bitrix24/handle.go index 1a6d4ffd..2c8ee702 100644 --- a/internal/channels/bitrix24/handle.go +++ b/internal/channels/bitrix24/handle.go @@ -13,6 +13,7 @@ import ( "github.com/nextlevelbuilder/goclaw/internal/channels" "github.com/nextlevelbuilder/goclaw/internal/channels/media" "github.com/nextlevelbuilder/goclaw/internal/store" + "github.com/nextlevelbuilder/goclaw/internal/systemmessages" ) // mentionMatcher is the compiled-once regex + rendered tag string used to @@ -91,8 +92,8 @@ func (c *Channel) DispatchEvent(ctx context.Context, evt *Event) { // 2. Group-only: require-mention gate + strip the mention from the text. // 3. Skip empty payloads (no text + no media). // 4. Policy: DM vs Group via BaseChannel.CheckDMPolicy / CheckGroupPolicy. -// Pairing policies trigger a pairing-reply stub (Phase 07 will wire up -// the full pairing flow; Phase 03 logs and drops). +// Pairing policies send a pairing reply (code + approve instructions) via +// sendPairingReply; deny drops silently. // 5. Build metadata map with bitrix_* keys so the agent can echo / reply // to the right dialog + message ID later. // 6. Forward to BaseChannel.HandleMessage → publishes bus.InboundMessage. @@ -258,16 +259,17 @@ func (c *Channel) handleMessage(ctx context.Context, evt *Event) { peerKind = "group" } - // Policy gating. BaseChannel reports PolicyNeedsPairing for unpaired DMs - // under the default "pairing" policy. Phase 07 will answer with a real - // pairing reply; Phase 03 logs the intent so the flow is visible while - // the rest of the channel comes online. + // Policy gating. BaseChannel reports PolicyNeedsPairing for unpaired senders + // under the "pairing" policy. Answer with a real pairing reply (code + how to + // approve) via the v2 chat API, matching the other channels (Discord/Zalo/ + // WhatsApp/Slack/Feishu). Group pairings are keyed by "group:" so the + // generated code matches what CheckGroupPolicy.IsPaired later verifies. if peerKind == "direct" { switch c.CheckDMPolicy(ctx, senderID, c.cfg.DMPolicy) { case channels.PolicyDeny: return case channels.PolicyNeedsPairing: - c.logPairingNeeded(senderID, chatID, peerKind) + c.sendPairingReply(ctx, senderID, chatID, peerKind) return } } else { @@ -275,7 +277,7 @@ func (c *Channel) handleMessage(ctx context.Context, evt *Event) { case channels.PolicyDeny: return case channels.PolicyNeedsPairing: - c.logPairingNeeded(senderID, chatID, peerKind) + c.sendPairingReply(ctx, "group:"+chatID, chatID, peerKind) return } } @@ -301,6 +303,11 @@ func (c *Channel) handleMessage(ctx context.Context, evt *Event) { // Bitrix replyId. MetaKeyMessageID: evt.Params.MessageID, MetaKeyVisibility: visibility, + // Generic run-tracking key the gateway consumer reads to thread + // streaming / status-reaction events back to this message (all other + // channels set it). Distinct from bitrix_message_id above (same value + // here, but that key is Bitrix-specific reply-link routing). + "message_id": evt.Params.MessageID, } // Echo the connector sender tag back on the reply so the Open Channel // connector routes the answer to the right external message: @@ -597,16 +604,45 @@ func (c *Channel) mention() *mentionMatcher { return c.mentionRe } -// logPairingNeeded is the Phase 03 stand-in for a real pairing reply. -// Phase 07 replaces this with a proper pairing message via imbot.message.add; -// until then we log the intent so operators can see unpaired attempts. -func (c *Channel) logPairingNeeded(senderID, chatID, peerKind string) { - if !c.CanSendPairingNotif(senderID, pairingDebounce) { +// sendPairingReply requests a pairing code for an unpaired sender and sends it +// back into the chat via the v2 API (imbot.v2.Chat.Message.send) so an admin can +// approve access. Mirrors the pairing reply the other channels already send +// (Discord/Zalo/WhatsApp/Slack/Feishu) using the shared pairing.account_required +// system message. Debounced per-sender so a spammy sender can't flood the chat. +// +// pairingSenderID is the id the pairing is keyed by — the raw senderID for DMs, +// "group:" for groups — so the generated code matches what CheckDMPolicy +// / CheckGroupPolicy later verify via IsPaired. +func (c *Channel) sendPairingReply(ctx context.Context, pairingSenderID, chatID, peerKind string) { + ps := c.PairingService() + if ps == nil { return } - c.MarkPairingNotifSent(senderID) - slog.Info("bitrix24: pairing required (Phase 07 will send pairing reply)", - "sender_id", senderID, "chat_id", chatID, "peer_kind", peerKind, + if !c.CanSendPairingNotif(pairingSenderID, pairingDebounce) { + return + } + + code, err := ps.RequestPairing(ctx, pairingSenderID, c.Name(), chatID, "default", nil) + if err != nil { + slog.Debug("bitrix24 pairing: request failed", + "sender_id", pairingSenderID, "chat_id", chatID, "err", err) + return + } + + replyText := c.SystemMessage("", systemmessages.KeyPairingAccountRequired, systemmessages.Vars{ + "platform": "Bitrix24", + "sender_id": pairingSenderID, + "code": code, + }) + + if err := c.sendChunkV2Public(ctx, chatID, replyText, 0); err != nil { + slog.Warn("bitrix24 pairing: reply send failed", + "sender_id", pairingSenderID, "chat_id", chatID, "err", err) + return + } + c.MarkPairingNotifSent(pairingSenderID) + slog.Info("bitrix24 pairing: reply sent", + "sender_id", pairingSenderID, "chat_id", chatID, "peer_kind", peerKind, "portal", c.cfg.Portal) } diff --git a/internal/channels/bitrix24/pairing_test.go b/internal/channels/bitrix24/pairing_test.go new file mode 100644 index 00000000..d7a99fbb --- /dev/null +++ b/internal/channels/bitrix24/pairing_test.go @@ -0,0 +1,118 @@ +package bitrix24 + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + + "github.com/nextlevelbuilder/goclaw/internal/store" + "github.com/nextlevelbuilder/goclaw/internal/systemmessages" +) + +// fakeBitrixPairingStore implements store.PairingStore for pairing-reply tests. +type fakeBitrixPairingStore struct { + requested bool + requestedFor string +} + +func (s *fakeBitrixPairingStore) RequestPairing(_ context.Context, senderID, _, _, _ string, _ map[string]string) (string, error) { + s.requested = true + s.requestedFor = senderID + return "code123", nil +} +func (s *fakeBitrixPairingStore) ApprovePairing(context.Context, string, string) (*store.PairedDeviceData, error) { + return nil, nil +} +func (s *fakeBitrixPairingStore) DenyPairing(context.Context, string) error { return nil } +func (s *fakeBitrixPairingStore) RevokePairing(context.Context, string, string) error { return nil } +func (s *fakeBitrixPairingStore) IsPaired(context.Context, string, string) (bool, error) { + return false, nil +} +func (s *fakeBitrixPairingStore) ListPending(context.Context) []store.PairingRequestData { return nil } +func (s *fakeBitrixPairingStore) ListPaired(context.Context) []store.PairedDeviceData { return nil } +func (s *fakeBitrixPairingStore) MigrateGroupChatID(context.Context, string, string, string) error { + return nil +} + +// TestSendPairingReply_SendsCodeViaV2 verifies an unpaired sender gets a pairing +// code sent back through the v2 chat API (imbot.v2.Chat.Message.send), not the +// old v1 imbot.message.add, and not silence. +func TestSendPairingReply_SendsCodeViaV2(t *testing.T) { + var calls int32 + var gotPath, gotMessage string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + atomic.AddInt32(&calls, 1) + gotPath = r.URL.Path + _ = r.ParseForm() + gotMessage = r.Form.Get("fields[message]") + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(map[string]any{"result": map[string]any{"id": 999}}) + })) + defer srv.Close() + + bc := newChannelWithBoundPortal(t, srv) + bc.SetSystemMessages(systemmessages.NewResolver(nil)) + ps := &fakeBitrixPairingStore{} + bc.SetPairingService(ps) + + bc.sendPairingReply(context.Background(), "42", "chat100", "direct") + + if atomic.LoadInt32(&calls) == 0 { + t.Fatal("expected a REST send for the pairing reply, got none (bot stayed silent)") + } + if !strings.HasSuffix(gotPath, "/rest/imbot.v2.Chat.Message.send.json") { + t.Errorf("pairing reply used wrong endpoint %q; want v2 imbot.v2.Chat.Message.send", gotPath) + } + if !ps.requested { + t.Error("expected RequestPairing to be called to generate a code") + } + if !strings.Contains(gotMessage, "code123") { + t.Errorf("pairing message should contain the code; got %q", gotMessage) + } +} + +// TestSendPairingReply_NoService_NoOp: without a pairing service the call is a +// safe no-op (no send). +func TestSendPairingReply_NoService_NoOp(t *testing.T) { + var calls int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + atomic.AddInt32(&calls, 1) + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(map[string]any{"result": map[string]any{"id": 1}}) + })) + defer srv.Close() + + bc := newChannelWithBoundPortal(t, srv) // no SetPairingService → PairingService() nil + bc.sendPairingReply(context.Background(), "42", "chat100", "direct") + + if got := atomic.LoadInt32(&calls); got != 0 { + t.Errorf("expected no send without a pairing service, got %d", got) + } +} + +// TestSendPairingReply_Debounced: a second call within the debounce window does +// not fire a second send. +func TestSendPairingReply_Debounced(t *testing.T) { + var calls int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + atomic.AddInt32(&calls, 1) + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(map[string]any{"result": map[string]any{"id": 1}}) + })) + defer srv.Close() + + bc := newChannelWithBoundPortal(t, srv) + bc.SetSystemMessages(systemmessages.NewResolver(nil)) + bc.SetPairingService(&fakeBitrixPairingStore{}) + + bc.sendPairingReply(context.Background(), "42", "chat100", "direct") + bc.sendPairingReply(context.Background(), "42", "chat100", "direct") // debounced + + if got := atomic.LoadInt32(&calls); got != 1 { + t.Errorf("expected 1 send (second debounced), got %d", got) + } +} diff --git a/internal/channels/bitrix24/reactions.go b/internal/channels/bitrix24/reactions.go new file mode 100644 index 00000000..85865997 --- /dev/null +++ b/internal/channels/bitrix24/reactions.go @@ -0,0 +1,167 @@ +package bitrix24 + +import ( + "context" + "errors" + "log/slog" + "strconv" + "strings" + "sync" + "time" + + "github.com/nextlevelbuilder/goclaw/internal/channels" +) + +// Compile-time guarantee the Bitrix24 channel provides status reactions. +var _ channels.ReactionChannel = (*Channel)(nil) + +// reactionDebounceInterval throttles intermediate status-reaction updates so a +// fast think→tool→tool sequence doesn't hammer the REST API. Terminal states +// (done/error) bypass the debounce so the final outcome always lands. +const reactionDebounceInterval = 700 * time.Millisecond + +// statusReaction maps a GoClaw agent status to a Bitrix24 v2 reaction code. +// Codes come from imbot.v2.Chat.Message.Reaction.add's documented set. +var statusReaction = map[string]string{ + "queued": "eyes", // 👀 waiting to process + "thinking": "thinkingFace", // 🤔 processing + "tool": "fire", // 🔥 running a tool + "coding": "fire", // 🔥 executing code + "web": "fire", // 🔥 browsing / API call + "done": "whiteHeavyCheckMark", // ✅ completed + "error": "crossMark", // ❌ failed + "stall": "sleepingSymbol", // 😴 no activity + "stallSoft": "sleepingSymbol", // 😴 + "stallHard": "flushedFace", // 😳 +} + +// reactionState tracks the current reaction code set on one user message so +// OnReactionEvent can replace it as the agent's status progresses. +type reactionState struct { + mu sync.Mutex + current string + lastUpdate time.Time +} + +// OnReactionEvent sets a status-reaction on the user's message to show agent +// progress (thinking → tool → done/error). Implements channels.ReactionChannel. +// messageID is the Bitrix MESSAGE_ID (numeric) of the message that started the run. +func (c *Channel) OnReactionEvent(ctx context.Context, chatID, messageID, status string) error { + level := c.cfg.ReactionLevel + if level == "" || level == "off" { + return nil + } + code, ok := statusReaction[status] + if !ok { + return nil + } + terminal := status == "done" || status == "error" + // "minimal" shows only the final outcome (done/error), not intermediate steps. + if level == "minimal" && !terminal { + return nil + } + + msgID, err := strconv.Atoi(strings.TrimSpace(messageID)) + if err != nil || msgID <= 0 { + return nil // not a Bitrix message id (e.g. connector synthetic id) + } + botID := c.BotID() + if botID <= 0 || c.Client() == nil { + return nil + } + + key := chatID + ":" + messageID + stVal, _ := c.reactions.LoadOrStore(key, &reactionState{}) + st := stVal.(*reactionState) + + st.mu.Lock() + defer st.mu.Unlock() + + if !terminal && time.Since(st.lastUpdate) < reactionDebounceInterval { + return nil + } + if st.current == code { + return nil + } + + // Bitrix has no "replace" — delete the previous reaction, then add the new. + if st.current != "" { + c.removeReaction(ctx, botID, msgID, st.current) + } + if err := c.addReaction(ctx, botID, msgID, code); err != nil { + slog.Debug("bitrix24 reaction: add failed", + "code", code, "message_id", msgID, "err", err) + return nil + } + st.current = code + st.lastUpdate = time.Now() + + // The run is over on terminal states — stop tracking this message. + if terminal { + c.reactions.Delete(key) + } + return nil +} + +// ClearReaction removes the current status-reaction from a message. +// Implements channels.ReactionChannel. +func (c *Channel) ClearReaction(ctx context.Context, chatID, messageID string) error { + key := chatID + ":" + messageID + stVal, ok := c.reactions.LoadAndDelete(key) + if !ok { + return nil + } + st := stVal.(*reactionState) + st.mu.Lock() + defer st.mu.Unlock() + if st.current == "" { + return nil + } + msgID, err := strconv.Atoi(strings.TrimSpace(messageID)) + if err != nil || msgID <= 0 { + return nil + } + if botID := c.BotID(); botID > 0 && c.Client() != nil { + c.removeReaction(ctx, botID, msgID, st.current) + } + st.current = "" + return nil +} + +// addReaction adds a bot reaction via imbot.v2.Chat.Message.Reaction.add. +// REACTION_ALREADY_SET is treated as success (the reaction is already present). +func (c *Channel) addReaction(ctx context.Context, botID, messageID int, code string) error { + _, err := c.Client().Call(ctx, "imbot.v2.Chat.Message.Reaction.add", map[string]any{ + "botId": botID, + "messageId": messageID, + "reaction": code, + }) + if err != nil && isReactionAlreadySet(err) { + return nil + } + return err +} + +// removeReaction removes a bot reaction via imbot.v2.Chat.Message.Reaction.delete. +// Best-effort: a failed delete is logged and swallowed (worst case a stale +// reaction lingers, which is cosmetic). +func (c *Channel) removeReaction(ctx context.Context, botID, messageID int, code string) { + if _, err := c.Client().Call(ctx, "imbot.v2.Chat.Message.Reaction.delete", map[string]any{ + "botId": botID, + "messageId": messageID, + "reaction": code, + }); err != nil { + slog.Debug("bitrix24 reaction: delete failed", + "code", code, "message_id", messageID, "err", err) + } +} + +// isReactionAlreadySet reports whether err is the benign REACTION_ALREADY_SET +// application error from Bitrix (the bot already reacted with this code). +func isReactionAlreadySet(err error) bool { + var apiErr *APIError + if errors.As(err, &apiErr) { + return apiErr.Code == "REACTION_ALREADY_SET" + } + return false +} diff --git a/internal/channels/bitrix24/reactions_test.go b/internal/channels/bitrix24/reactions_test.go new file mode 100644 index 00000000..3eef5585 --- /dev/null +++ b/internal/channels/bitrix24/reactions_test.go @@ -0,0 +1,129 @@ +package bitrix24 + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" +) + +type reactionCall struct { + op string // "add" | "delete" + reaction string +} + +// reactionRecorder serves the v2 reaction endpoints and records each call. +func reactionRecorder(t *testing.T) (*httptest.Server, func() []reactionCall) { + t.Helper() + var mu sync.Mutex + var calls []reactionCall + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _ = r.ParseForm() + op := "" + switch { + case strings.HasSuffix(r.URL.Path, "/rest/imbot.v2.Chat.Message.Reaction.add.json"): + op = "add" + case strings.HasSuffix(r.URL.Path, "/rest/imbot.v2.Chat.Message.Reaction.delete.json"): + op = "delete" + default: + t.Errorf("unexpected reaction path: %q", r.URL.Path) + } + mu.Lock() + calls = append(calls, reactionCall{op: op, reaction: r.Form.Get("reaction")}) + mu.Unlock() + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(map[string]any{"result": true}) + })) + return srv, func() []reactionCall { + mu.Lock() + defer mu.Unlock() + out := make([]reactionCall, len(calls)) + copy(out, calls) + return out + } +} + +func TestOnReactionEvent_FullFlow_AddThenReplaceViaV2(t *testing.T) { + srv, snapshot := reactionRecorder(t) + defer srv.Close() + bc := newChannelWithBoundPortal(t, srv) + bc.cfg.ReactionLevel = "full" + ctx := context.Background() + + // thinking → add thinkingFace + _ = bc.OnReactionEvent(ctx, "chat100", "5001", "thinking") + // done (terminal, bypasses debounce) → delete thinkingFace + add whiteHeavyCheckMark + _ = bc.OnReactionEvent(ctx, "chat100", "5001", "done") + + calls := snapshot() + if len(calls) != 3 { + t.Fatalf("expected 3 reaction calls (add, delete, add), got %d: %+v", len(calls), calls) + } + if calls[0].op != "add" || calls[0].reaction != "thinkingFace" { + t.Errorf("call[0] = %+v; want add thinkingFace", calls[0]) + } + if calls[1].op != "delete" || calls[1].reaction != "thinkingFace" { + t.Errorf("call[1] = %+v; want delete thinkingFace", calls[1]) + } + if calls[2].op != "add" || calls[2].reaction != "whiteHeavyCheckMark" { + t.Errorf("call[2] = %+v; want add whiteHeavyCheckMark", calls[2]) + } +} + +func TestOnReactionEvent_MinimalSkipsIntermediate(t *testing.T) { + srv, snapshot := reactionRecorder(t) + defer srv.Close() + bc := newChannelWithBoundPortal(t, srv) + bc.cfg.ReactionLevel = "minimal" + ctx := context.Background() + + _ = bc.OnReactionEvent(ctx, "chat100", "5001", "thinking") // skipped in minimal + _ = bc.OnReactionEvent(ctx, "chat100", "5001", "error") // terminal → shown + + calls := snapshot() + if len(calls) != 1 || calls[0].op != "add" || calls[0].reaction != "crossMark" { + t.Fatalf("minimal should send only the terminal reaction; got %+v", calls) + } +} + +func TestOnReactionEvent_OffAndInvalidAreNoOp(t *testing.T) { + srv, snapshot := reactionRecorder(t) + defer srv.Close() + + // off level + bc := newChannelWithBoundPortal(t, srv) + bc.cfg.ReactionLevel = "off" + _ = bc.OnReactionEvent(context.Background(), "chat100", "5001", "done") + + // full level but non-numeric message id (connector synthetic id) + bc2 := newChannelWithBoundPortal(t, srv) + bc2.cfg.ReactionLevel = "full" + _ = bc2.OnReactionEvent(context.Background(), "chat100", "not-a-number", "done") + + // full level, unknown status + bc2.cfg.ReactionLevel = "full" + _ = bc2.OnReactionEvent(context.Background(), "chat100", "5001", "bogus-status") + + if calls := snapshot(); len(calls) != 0 { + t.Fatalf("expected no reaction calls for off/invalid/unknown, got %+v", calls) + } +} + +func TestClearReaction_RemovesCurrent(t *testing.T) { + srv, snapshot := reactionRecorder(t) + defer srv.Close() + bc := newChannelWithBoundPortal(t, srv) + bc.cfg.ReactionLevel = "full" + ctx := context.Background() + + _ = bc.OnReactionEvent(ctx, "chat100", "5001", "thinking") // add thinkingFace + _ = bc.ClearReaction(ctx, "chat100", "5001") // delete thinkingFace + + calls := snapshot() + if len(calls) != 2 || calls[1].op != "delete" || calls[1].reaction != "thinkingFace" { + t.Fatalf("expected add then delete thinkingFace; got %+v", calls) + } +} diff --git a/internal/channels/routing_metadata.go b/internal/channels/routing_metadata.go index ce448596..a5bf5222 100644 --- a/internal/channels/routing_metadata.go +++ b/internal/channels/routing_metadata.go @@ -16,6 +16,13 @@ var routingMetaKeys = []string{ "pancake_mode", // pancake inbox vs comment routing "post_id", // pancake: post id for template vars "display_name", // pancake: commenter display name for template vars + // bitrix24 whisper/public visibility. MUST survive into intermediate + // replies (quick-ack, progress, reasoning bubbles, block replies) so a + // whisper-mode turn doesn't leak its FIRST (intermediate) message to the + // external connector while only the final answer is whispered. Literal + // string mirrors bitrix24.MetaKeyVisibility (kept as a string here to + // avoid a channels→bitrix24 import cycle). Non-bitrix channels ignore it. + "bitrix_visibility", } var finalReplyMetaKeys = append([]string{ diff --git a/internal/channels/routing_metadata_test.go b/internal/channels/routing_metadata_test.go index 1a668fa3..5c9aaab4 100644 --- a/internal/channels/routing_metadata_test.go +++ b/internal/channels/routing_metadata_test.go @@ -62,3 +62,20 @@ func TestCopyRoutingMeta_PreservesPancakePrivateReplyKeys(t *testing.T) { } } } + +// TestCopyRoutingMeta_PreservesBitrixVisibility guards the whisper-leak +// regression: intermediate replies (quick-ack, progress, reasoning bubbles, +// block replies) must carry bitrix_visibility so a whisper-mode turn does not +// leak its FIRST (intermediate) message to the external connector while only +// the final answer is whispered. copyRoutingMeta is the path used by every +// intermediate publish site; CopyFinalRoutingMeta must keep it too. +func TestCopyRoutingMeta_PreservesBitrixVisibility(t *testing.T) { + src := map[string]string{"bitrix_visibility": "whisper"} + + if got := copyRoutingMeta(src); got["bitrix_visibility"] != "whisper" { + t.Fatalf("copyRoutingMeta()[%q] = %q, want %q", "bitrix_visibility", got["bitrix_visibility"], "whisper") + } + if got := CopyFinalRoutingMeta(src); got["bitrix_visibility"] != "whisper" { + t.Fatalf("CopyFinalRoutingMeta()[%q] = %q, want %q", "bitrix_visibility", got["bitrix_visibility"], "whisper") + } +} diff --git a/internal/config/config_channels.go b/internal/config/config_channels.go index b3855436..c87ee3d4 100644 --- a/internal/config/config_channels.go +++ b/internal/config/config_channels.go @@ -576,16 +576,33 @@ type SessionsConfig struct { // TtsConfig configures text-to-speech. // Matching TS src/config/types.tts.ts. type TtsConfig struct { - Provider string `json:"provider,omitempty"` // "openai", "elevenlabs", "edge", "minimax", "gemini" - Auto string `json:"auto,omitempty"` // "off" (default), "always", "inbound", "tagged" - Mode string `json:"mode,omitempty"` // "final" (default), "all" - MaxLength int `json:"max_length,omitempty"` // max text length before truncation (default 1500) - TimeoutMs int `json:"timeout_ms,omitempty"` // API timeout in ms (default 30000) - OpenAI TtsOpenAIConfig `json:"openai"` - ElevenLabs TtsElevenLabsConfig `json:"elevenlabs"` - Edge TtsEdgeConfig `json:"edge"` - MiniMax TtsMiniMaxConfig `json:"minimax"` - Gemini TtsGeminiConfig `json:"gemini"` + Provider string `json:"provider,omitempty"` // "openai", "openai_compat", "elevenlabs", "edge", "minimax", "gemini" + Auto string `json:"auto,omitempty"` // "off" (default), "always", "inbound", "tagged" + Mode string `json:"mode,omitempty"` // "final" (default), "all" + MaxLength int `json:"max_length,omitempty"` // max text length before truncation (default 1500) + TimeoutMs int `json:"timeout_ms,omitempty"` // API timeout in ms (default 30000) + OpenAI TtsOpenAIConfig `json:"openai"` + OpenAICompat TtsOpenAICompatConfig `json:"openai_compat"` + ElevenLabs TtsElevenLabsConfig `json:"elevenlabs"` + Edge TtsEdgeConfig `json:"edge"` + MiniMax TtsMiniMaxConfig `json:"minimax"` + Gemini TtsGeminiConfig `json:"gemini"` +} + +// TtsOpenAICompatConfig configures a self-hosted OpenAI-compatible audio +// endpoint (gpu-manager, Speaches, vLLM, LocalAI, llama.cpp, Ollama) for both +// TTS and STT. Kept separate from TtsOpenAIConfig because the voice namespace +// is the engine's own — see internal/audio/openaicompat for why the provider +// name must not be "openai". +// +// Setting api_base is what enables the provider; there is no vendor default. +type TtsOpenAICompatConfig struct { + APIBase string `json:"api_base,omitempty"` // required to enable, e.g. "http://gpu-manager:8080/v1" + APIKey string `json:"api_key,omitempty"` // optional; omitted entirely when empty + Model string `json:"model,omitempty"` // TTS model; optional when the engine picks it from the voice + Voice string `json:"voice,omitempty"` // engine-specific voice ID, e.g. "fr_FR-gilles-low" + Format string `json:"format,omitempty"` // response_format; default "mp3", use "wav" for engines without an encoder + STTModel string `json:"stt_model,omitempty"` // transcription model; optional } // TtsGeminiConfig configures the Google Gemini TTS provider. diff --git a/internal/store/pg/teams_tasks_followup.go b/internal/store/pg/teams_tasks_followup.go index 72bf8a6c..2087839e 100644 --- a/internal/store/pg/teams_tasks_followup.go +++ b/internal/store/pg/teams_tasks_followup.go @@ -52,7 +52,7 @@ func (s *PGTeamStore) ListAllFollowupDueTasks(ctx context.Context) ([]store.Team rows, err := s.db.QueryContext(ctx, `SELECT `+taskSelectCols+` `+taskJoinClause+` - `+v2ActiveTeamJoin+` + `+activeTeamJoin+` WHERE t.followup_at IS NOT NULL AND t.followup_at <= $1 AND t.status = $2 diff --git a/internal/store/pg/teams_tasks_progress.go b/internal/store/pg/teams_tasks_progress.go index fcd172e3..d6c2c9ef 100644 --- a/internal/store/pg/teams_tasks_progress.go +++ b/internal/store/pg/teams_tasks_progress.go @@ -71,15 +71,14 @@ func (s *PGTeamStore) RenewTaskLock(ctx context.Context, taskID, teamID uuid.UUI } // ============================================================ -// Stale recovery (batch — all v2 active teams in one query) +// Stale recovery (batch — all active teams in one query) // ============================================================ -// v2ActiveTeamJoin is the JOIN clause that filters to v2 active teams. -const v2ActiveTeamJoin = `JOIN agent_teams tm ON tm.id = t.team_id - AND tm.status = 'active' - AND COALESCE((tm.settings->>'version')::int, 0) >= 2` +// activeTeamJoin is the JOIN clause that filters to active teams. +const activeTeamJoin = `JOIN agent_teams tm ON tm.id = t.team_id + AND tm.status = 'active'` -// RecoverAllStaleTasks resets in_progress tasks with expired locks across all v2 active teams. +// RecoverAllStaleTasks resets in_progress tasks with expired locks across all active teams. func (s *PGTeamStore) RecoverAllStaleTasks(ctx context.Context) ([]store.RecoveredTaskInfo, error) { now := time.Now() rows, err := s.db.QueryContext(ctx, @@ -89,7 +88,6 @@ func (s *PGTeamStore) RecoverAllStaleTasks(ctx context.Context) ([]store.Recover followup_channel = NULL, followup_chat_id = NULL, updated_at = $2 FROM agent_teams tm WHERE t.team_id = tm.id AND tm.status = 'active' - AND COALESCE((tm.settings->>'version')::int, 0) >= 2 AND t.status = $3 AND t.lock_expires_at IS NOT NULL AND t.lock_expires_at < $2 RETURNING t.id, t.team_id, t.tenant_id, t.task_number, t.subject, COALESCE(t.channel, ''), COALESCE(t.chat_id, '')`, @@ -102,7 +100,7 @@ func (s *PGTeamStore) RecoverAllStaleTasks(ctx context.Context) ([]store.Recover return scanRecoveredTaskInfoRows(rows) } -// ForceRecoverAllTasks resets ALL in_progress tasks across v2 active teams (startup). +// ForceRecoverAllTasks resets ALL in_progress tasks across active teams (startup). func (s *PGTeamStore) ForceRecoverAllTasks(ctx context.Context) ([]store.RecoveredTaskInfo, error) { now := time.Now() rows, err := s.db.QueryContext(ctx, @@ -112,7 +110,6 @@ func (s *PGTeamStore) ForceRecoverAllTasks(ctx context.Context) ([]store.Recover followup_channel = NULL, followup_chat_id = NULL, updated_at = $2 FROM agent_teams tm WHERE t.team_id = tm.id AND tm.status = 'active' - AND COALESCE((tm.settings->>'version')::int, 0) >= 2 AND t.status = $3 RETURNING t.id, t.team_id, t.tenant_id, t.task_number, t.subject, COALESCE(t.channel, ''), COALESCE(t.chat_id, '')`, store.TeamTaskStatusPending, now, store.TeamTaskStatusInProgress, @@ -148,7 +145,7 @@ func (s *PGTeamStore) ListRecoverableTasks(ctx context.Context, teamID uuid.UUID return scanTaskRowsJoined(rows) } -// MarkAllStaleTasks marks pending tasks older than olderThan as stale across all v2 active teams. +// MarkAllStaleTasks marks pending tasks older than olderThan as stale across all active teams. func (s *PGTeamStore) MarkAllStaleTasks(ctx context.Context, olderThan time.Time) ([]store.RecoveredTaskInfo, error) { now := time.Now() rows, err := s.db.QueryContext(ctx, @@ -156,7 +153,6 @@ func (s *PGTeamStore) MarkAllStaleTasks(ctx context.Context, olderThan time.Time SET status = $1, updated_at = $2 FROM agent_teams tm WHERE t.team_id = tm.id AND tm.status = 'active' - AND COALESCE((tm.settings->>'version')::int, 0) >= 2 AND t.status = $3 AND t.updated_at < $4 RETURNING t.id, t.team_id, t.tenant_id, t.task_number, t.subject, COALESCE(t.channel, ''), COALESCE(t.chat_id, '')`, store.TeamTaskStatusStale, now, store.TeamTaskStatusPending, olderThan, @@ -168,7 +164,7 @@ func (s *PGTeamStore) MarkAllStaleTasks(ctx context.Context, olderThan time.Time return scanRecoveredTaskInfoRows(rows) } -// MarkInReviewStaleTasks marks in_review tasks older than olderThan as stale across all v2 active teams. +// MarkInReviewStaleTasks marks in_review tasks older than olderThan as stale across all active teams. func (s *PGTeamStore) MarkInReviewStaleTasks(ctx context.Context, olderThan time.Time) ([]store.RecoveredTaskInfo, error) { now := time.Now() rows, err := s.db.QueryContext(ctx, @@ -176,7 +172,6 @@ func (s *PGTeamStore) MarkInReviewStaleTasks(ctx context.Context, olderThan time SET status = $1, updated_at = $2 FROM agent_teams tm WHERE t.team_id = tm.id AND tm.status = 'active' - AND COALESCE((tm.settings->>'version')::int, 0) >= 2 AND t.status = $3 AND t.updated_at < $4 RETURNING t.id, t.team_id, t.tenant_id, t.task_number, t.subject, COALESCE(t.channel, ''), COALESCE(t.chat_id, '')`, store.TeamTaskStatusStale, now, store.TeamTaskStatusInReview, olderThan, @@ -197,7 +192,6 @@ func (s *PGTeamStore) FixOrphanedBlockedTasks(ctx context.Context) ([]store.Reco SET blocked_by = '{}', status = $1, updated_at = $2 FROM agent_teams tm WHERE t.team_id = tm.id AND tm.status = 'active' - AND COALESCE((tm.settings->>'version')::int, 0) >= 2 AND t.status = 'blocked' AND array_length(t.blocked_by, 1) > 0 AND NOT EXISTS ( diff --git a/internal/store/sqlitestore/teams_tasks_activity.go b/internal/store/sqlitestore/teams_tasks_activity.go index c76f1dfa..90ecf76e 100644 --- a/internal/store/sqlitestore/teams_tasks_activity.go +++ b/internal/store/sqlitestore/teams_tasks_activity.go @@ -352,11 +352,10 @@ func (s *SQLiteTeamStore) RenewTaskLock(ctx context.Context, taskID, teamID uuid // Stale recovery // ============================================================ -// v2ActiveTeamJoin is the JOIN that filters to v2 active teams. +// activeTeamJoin is the JOIN that filters to active teams. // SQLite uses json_extract instead of PG's ->> operator. -const v2ActiveTeamJoin = `JOIN agent_teams tm ON tm.id = t.team_id - AND tm.status = 'active' - AND COALESCE(CAST(json_extract(tm.settings, '$.version') AS INTEGER), 0) >= 2` +const activeTeamJoin = `JOIN agent_teams tm ON tm.id = t.team_id + AND tm.status = 'active'` func (s *SQLiteTeamStore) RecoverAllStaleTasks(ctx context.Context) ([]store.RecoveredTaskInfo, error) { now := time.Now() @@ -368,7 +367,6 @@ func (s *SQLiteTeamStore) RecoverAllStaleTasks(ctx context.Context) ([]store.Rec WHERE status = ? AND lock_expires_at IS NOT NULL AND lock_expires_at < ? AND team_id IN ( SELECT id FROM agent_teams WHERE status = 'active' - AND COALESCE(CAST(json_extract(settings, '$.version') AS INTEGER), 0) >= 2 )`, store.TeamTaskStatusPending, now, store.TeamTaskStatusInProgress, now, ) @@ -387,7 +385,6 @@ func (s *SQLiteTeamStore) ForceRecoverAllTasks(ctx context.Context) ([]store.Rec WHERE status = ? AND team_id IN ( SELECT id FROM agent_teams WHERE status = 'active' - AND COALESCE(CAST(json_extract(settings, '$.version') AS INTEGER), 0) >= 2 )`, store.TeamTaskStatusPending, now, store.TeamTaskStatusInProgress, ) @@ -426,7 +423,6 @@ func (s *SQLiteTeamStore) MarkAllStaleTasks(ctx context.Context, olderThan time. WHERE status = ? AND updated_at < ? AND team_id IN ( SELECT id FROM agent_teams WHERE status = 'active' - AND COALESCE(CAST(json_extract(settings, '$.version') AS INTEGER), 0) >= 2 )`, store.TeamTaskStatusStale, now, store.TeamTaskStatusPending, olderThan, ) @@ -443,7 +439,6 @@ func (s *SQLiteTeamStore) MarkInReviewStaleTasks(ctx context.Context, olderThan WHERE status = ? AND updated_at < ? AND team_id IN ( SELECT id FROM agent_teams WHERE status = 'active' - AND COALESCE(CAST(json_extract(settings, '$.version') AS INTEGER), 0) >= 2 )`, store.TeamTaskStatusStale, now, store.TeamTaskStatusInReview, olderThan, ) @@ -462,7 +457,6 @@ func (s *SQLiteTeamStore) FixOrphanedBlockedTasks(ctx context.Context) ([]store. COALESCE(t.channel, ''), COALESCE(t.chat_id, '') FROM team_tasks t JOIN agent_teams tm ON tm.id = t.team_id AND tm.status = 'active' - AND COALESCE(CAST(json_extract(tm.settings, '$.version') AS INTEGER), 0) >= 2 WHERE t.status = 'blocked' AND t.blocked_by IS NOT NULL AND t.blocked_by != '[]' AND NOT EXISTS ( @@ -501,7 +495,6 @@ func (s *SQLiteTeamStore) queryRecoveredTasks(ctx context.Context, status string AND t.updated_at >= ? AND t.team_id IN ( SELECT id FROM agent_teams WHERE status = 'active' - AND COALESCE(CAST(json_extract(settings, '$.version') AS INTEGER), 0) >= 2 ) ORDER BY t.updated_at DESC LIMIT 200`, @@ -599,7 +592,7 @@ func (s *SQLiteTeamStore) ListAllFollowupDueTasks(ctx context.Context) ([]store. rows, err := s.db.QueryContext(ctx, `SELECT `+taskSelectCols+` `+taskJoinClause+` - `+v2ActiveTeamJoin+` + `+activeTeamJoin+` WHERE t.followup_at IS NOT NULL AND t.followup_at <= ? AND t.status = ? diff --git a/internal/tools/tts.go b/internal/tools/tts.go index 2f101efc..d0eb097c 100644 --- a/internal/tools/tts.go +++ b/internal/tools/tts.go @@ -27,13 +27,20 @@ import ( // Implements Tool + ContextualTool interfaces. // Per-call channel is read from ctx for thread-safety. type TtsTool struct { - mu sync.RWMutex - manager *tts.Manager - vaultIntc *VaultInterceptor + mu sync.RWMutex + manager *tts.Manager + vaultIntc *VaultInterceptor + systemConfigs store.SystemConfigStore } func (t *TtsTool) SetVaultInterceptor(v *VaultInterceptor) { t.vaultIntc = v } +func (t *TtsTool) SetSystemConfigStore(s store.SystemConfigStore) { + t.mu.Lock() + defer t.mu.Unlock() + t.systemConfigs = s +} + // NewTtsTool creates a TTS tool backed by the given manager. func NewTtsTool(mgr *tts.Manager) *TtsTool { return &TtsTool{manager: mgr} @@ -98,7 +105,7 @@ type agentAudioConfig struct { // resolveVoiceAndModel computes the effective voice + model IDs for the // request using the documented precedence order: // -// args > agent (store.AgentAudioFromCtx OtherConfig) > tenant (BuiltinToolSettings) > empty. +// args > agent (store.AgentAudioFromCtx OtherConfig) > tenant (BuiltinToolSettings) > system_configs > empty. // // Empty return values signal "use provider default" downstream — they are not // errors. Missing agent snapshot emits slog.Warn so operators can spot @@ -108,7 +115,7 @@ type agentAudioConfig struct { // tts_voice_id override (not from tool args or tenant defaults). Callers use // this flag to emit a warning when the voice is later found incompatible with // the selected provider. -func (t *TtsTool) resolveVoiceAndModel(ctx context.Context, argVoice, argModel string) (voice, model string, voiceFromAgent bool) { +func (t *TtsTool) resolveVoiceAndModel(ctx context.Context, providerName, argVoice, argModel string) (voice, model string, voiceFromAgent bool) { voice, model = argVoice, argModel // Pull agent-level config from the dispatcher-injected snapshot. @@ -153,6 +160,22 @@ func (t *TtsTool) resolveVoiceAndModel(ctx context.Context, argVoice, argModel s model = tenantCfg.DefaultModel } } + + t.mu.RLock() + systemConfigs := t.systemConfigs + t.mu.RUnlock() + if (voice == "" || model == "") && systemConfigs != nil && providerName != "" { + if voice == "" { + if v, err := systemConfigs.Get(ctx, "tts."+providerName+".voice"); err == nil && v != "" { + voice = v + } + } + if model == "" { + if m, err := systemConfigs.Get(ctx, "tts."+providerName+".model"); err == nil && m != "" { + model = m + } + } + } return voice, model, voiceFromAgent } @@ -252,9 +275,6 @@ func (t *TtsTool) Execute(ctx context.Context, args map[string]any) *Result { argModel, _ := args["model"].(string) providerName, _ := args["provider"].(string) - // Resolve voice/model via args > agent (ctx snapshot) > tenant > default. - voice, model, voiceFromAgent := t.resolveVoiceAndModel(ctx, argVoice, argModel) - // Read generic agent TTS params once; adapt PER-ATTEMPT below (Finding #1 CRITICAL). // Storing generic keys here so each fallback provider gets its own adapted copy. genericAgentParams := t.resolveAgentGenericTTSParams(ctx) @@ -264,6 +284,12 @@ func (t *TtsTool) Execute(ctx context.Context, args map[string]any) *Result { mgr := t.manager t.mu.RUnlock() + effectiveProvider := providerName + if effectiveProvider == "" { + effectiveProvider = t.resolvePrimary(ctx, mgr) + } + voice, model, voiceFromAgent := t.resolveVoiceAndModel(ctx, effectiveProvider, argVoice, argModel) + // Determine format based on channel (read from ctx — thread-safe) channel := ToolChannelFromCtx(ctx) opts := tts.Options{Voice: voice, Model: model} diff --git a/internal/tools/tts_agent_ctx_test.go b/internal/tools/tts_agent_ctx_test.go index 53c66a84..bb943518 100644 --- a/internal/tools/tts_agent_ctx_test.go +++ b/internal/tools/tts_agent_ctx_test.go @@ -35,7 +35,7 @@ func ctxWithAgentAudio(t *testing.T, voiceID, modelID string) context.Context { func TestResolveVoiceAndModel_ArgsWinOverAgent(t *testing.T) { tool := NewTtsTool(makeTTSManager("elevenlabs")) ctx := ctxWithAgentAudio(t, "AGENT_V", "AGENT_M") - v, m, _ := tool.resolveVoiceAndModel(ctx, "ARG_V", "ARG_M") + v, m, _ := tool.resolveVoiceAndModel(ctx, "edge", "ARG_V", "ARG_M") if v != "ARG_V" { t.Errorf("voice: got %q, want ARG_V (args must win)", v) } @@ -50,7 +50,7 @@ func TestResolveVoiceAndModel_AgentWinsOverTenantWhenArgsEmpty(t *testing.T) { ctx = WithBuiltinToolSettings(ctx, BuiltinToolSettings{ "tts": rawJSON(t, map[string]string{"default_voice_id": "TENANT_V", "default_model": "TENANT_M"}), }) - v, m, _ := tool.resolveVoiceAndModel(ctx, "", "") + v, m, _ := tool.resolveVoiceAndModel(ctx, "edge", "", "") if v != "AGENT_V" { t.Errorf("voice: got %q, want AGENT_V (agent > tenant)", v) } @@ -65,7 +65,7 @@ func TestResolveVoiceAndModel_TenantFallbackWhenAgentSilent(t *testing.T) { ctx := WithBuiltinToolSettings(context.Background(), BuiltinToolSettings{ "tts": rawJSON(t, map[string]string{"default_voice_id": "TENANT_V", "default_model": "TENANT_M"}), }) - v, m, _ := tool.resolveVoiceAndModel(ctx, "", "") + v, m, _ := tool.resolveVoiceAndModel(ctx, "edge", "", "") if v != "TENANT_V" { t.Errorf("voice: got %q, want TENANT_V", v) } @@ -76,7 +76,7 @@ func TestResolveVoiceAndModel_TenantFallbackWhenAgentSilent(t *testing.T) { func TestResolveVoiceAndModel_EmptyAllMeansDefault(t *testing.T) { tool := NewTtsTool(makeTTSManager("elevenlabs")) - v, m, _ := tool.resolveVoiceAndModel(context.Background(), "", "") + v, m, _ := tool.resolveVoiceAndModel(context.Background(), "edge", "", "") if v != "" { t.Errorf("voice: got %q, want empty (no sources → provider default)", v) } @@ -92,7 +92,7 @@ func TestResolveVoiceAndModel_PartialAgentConfig(t *testing.T) { ctx = WithBuiltinToolSettings(ctx, BuiltinToolSettings{ "tts": rawJSON(t, map[string]string{"default_model": "TENANT_M"}), }) - v, m, _ := tool.resolveVoiceAndModel(ctx, "", "") + v, m, _ := tool.resolveVoiceAndModel(ctx, "edge", "", "") if v != "AGENT_V" { t.Errorf("voice: got %q, want AGENT_V", v) } diff --git a/internal/tools/tts_systemconfigs_fallback_test.go b/internal/tools/tts_systemconfigs_fallback_test.go new file mode 100644 index 00000000..3fbbf553 --- /dev/null +++ b/internal/tools/tts_systemconfigs_fallback_test.go @@ -0,0 +1,76 @@ +package tools + +import ( + "context" + "testing" +) + +type fakeSystemConfigStore struct { + data map[string]string +} + +func (f *fakeSystemConfigStore) Get(_ context.Context, key string) (string, error) { + return f.data[key], nil +} +func (f *fakeSystemConfigStore) Set(_ context.Context, key, value string) error { + if f.data == nil { + f.data = map[string]string{} + } + f.data[key] = value + return nil +} +func (f *fakeSystemConfigStore) Delete(_ context.Context, key string) error { + delete(f.data, key) + return nil +} +func (f *fakeSystemConfigStore) List(_ context.Context) (map[string]string, error) { + return f.data, nil +} + +func TestResolveVoiceAndModel_SystemConfigsFallback(t *testing.T) { + tool := NewTtsTool(nil) + sc := &fakeSystemConfigStore{data: map[string]string{ + "tts.edge.voice": "vi-VN-HoaiMyNeural", + "tts.edge.model": "edge-tts-1", + }} + tool.SetSystemConfigStore(sc) + + v, m, _ := tool.resolveVoiceAndModel(context.Background(), "edge", "", "") + if v != "vi-VN-HoaiMyNeural" { + t.Errorf("voice fallback failed: got %q, want vi-VN-HoaiMyNeural", v) + } + if m != "edge-tts-1" { + t.Errorf("model fallback failed: got %q, want edge-tts-1", m) + } +} + +func TestResolveVoiceAndModel_ArgWinsOverSystemConfigs(t *testing.T) { + tool := NewTtsTool(nil) + tool.SetSystemConfigStore(&fakeSystemConfigStore{data: map[string]string{ + "tts.edge.voice": "vi-VN-HoaiMyNeural", + }}) + + v, _, _ := tool.resolveVoiceAndModel(context.Background(), "edge", "en-US-AriaNeural", "") + if v != "en-US-AriaNeural" { + t.Errorf("arg voice must win over system_configs: got %q", v) + } +} + +func TestResolveVoiceAndModel_NoStoreNoFallback(t *testing.T) { + tool := NewTtsTool(nil) + v, m, _ := tool.resolveVoiceAndModel(context.Background(), "edge", "", "") + if v != "" || m != "" { + t.Errorf("expected empty fallback when no system_configs wired, got voice=%q model=%q", v, m) + } +} + +func TestResolveVoiceAndModel_EmptyProviderSkipsFallback(t *testing.T) { + tool := NewTtsTool(nil) + tool.SetSystemConfigStore(&fakeSystemConfigStore{data: map[string]string{ + "tts..voice": "should-not-match", + }}) + v, m, _ := tool.resolveVoiceAndModel(context.Background(), "", "", "") + if v != "" || m != "" { + t.Errorf("empty provider must skip lookup, got voice=%q model=%q", v, m) + } +} diff --git a/ui/web/src/pages/config/sections/system-messages-section.tsx b/ui/web/src/pages/config/sections/system-messages-section.tsx index 3612034b..aa4ea8d4 100644 --- a/ui/web/src/pages/config/sections/system-messages-section.tsx +++ b/ui/web/src/pages/config/sections/system-messages-section.tsx @@ -30,7 +30,10 @@ export function SystemMessagesSection({ data, schema, onSave, saving }: Props) { const definitions = useMemo(() => systemMessageDefinitionsFromSchema(schema), [schema]); const [draft, setDraft] = useState(() => normalizeSystemMessagesDraft(data, definitions)); const [selectedKey, setSelectedKey] = useState(""); - const [selectedLocale, setSelectedLocale] = useState<(typeof SYSTEM_MESSAGE_LOCALES)[number]["code"]>("vi"); + const initialLocale = (i18n.resolvedLanguage || i18n.language || "en").slice(0, 2) as (typeof SYSTEM_MESSAGE_LOCALES)[number]["code"]; + const [selectedLocale, setSelectedLocale] = useState<(typeof SYSTEM_MESSAGE_LOCALES)[number]["code"]>( + SYSTEM_MESSAGE_LOCALES.some((l) => l.code === initialLocale) ? initialLocale : "en", + ); const [defaultLocale, setDefaultLocale] = useState<(typeof SYSTEM_MESSAGE_LOCALES)[number]["code"]>( (data?.default_locale as (typeof SYSTEM_MESSAGE_LOCALES)[number]["code"]) || "en", );