diff --git a/agent/cursor/session.go b/agent/cursor/session.go index cb84215805..dcf415e1c2 100644 --- a/agent/cursor/session.go +++ b/agent/cursor/session.go @@ -362,7 +362,13 @@ func (cs *cursorSession) handleInteractionQuery(raw map[string]any) { } // Store pending query so RespondPermission can write the right response. + // If a previous query is still pending (unlikely—Cursor blocks waiting for + // a response—but possible under network lag), deny it first to unblock the + // CLI before accepting the new one. cs.pendingMu.Lock() + if prev := cs.pending; prev != nil { + cs.writeInteractionResponse(prev.id, prev.queryType, false, "superseded by new permission request") + } cs.pending = &pendingInteractionQuery{id: queryID, queryType: queryType} cs.pendingMu.Unlock() diff --git a/cmd/cc-connect/main.go b/cmd/cc-connect/main.go index 1d3290f5b7..c7fbff45c4 100644 --- a/cmd/cc-connect/main.go +++ b/cmd/cc-connect/main.go @@ -550,6 +550,10 @@ func main() { // Wire shell configuration shell, shellFlag, shellProfile := config.EffectiveShell(cfg, &proj) + if err := config.ValidateShellBinary(shell); err != nil { + slog.Warn("invalid shell configuration, falling back to default", "project", proj.Name, "error", err) + shell, shellFlag, shellProfile = config.EffectiveShell(&config.Config{}, nil) + } engine.SetShell(shell, shellFlag, shellProfile) // Wire hooks @@ -966,14 +970,23 @@ func main() { slog.Warn("timer store unavailable", "error", err) } var timerSched *core.TimerScheduler - if timerStore != nil { + timerEnabled := cfg.Timer.Enabled == nil || *cfg.Timer.Enabled + if timerStore != nil && timerEnabled { timerSched = core.NewTimerScheduler(timerStore) - if cfg.Cron.Silent != nil && *cfg.Cron.Silent { + // Timer-specific silent/session_mode, fallback to cron settings for backward compat + if cfg.Timer.Silent != nil { + timerSched.SetDefaultSilent(*cfg.Timer.Silent) + } else if cfg.Cron.Silent != nil && *cfg.Cron.Silent { timerSched.SetDefaultSilent(true) } - if cfg.Cron.SessionMode != "" { + if cfg.Timer.SessionMode != "" { + timerSched.SetDefaultSessionMode(cfg.Timer.SessionMode) + } else if cfg.Cron.SessionMode != "" { timerSched.SetDefaultSessionMode(cfg.Cron.SessionMode) } + if cfg.Timer.MaxPendingJobs > 0 { + timerSched.SetMaxPendingJobs(cfg.Timer.MaxPendingJobs) + } for i, e := range engines { timerSched.RegisterEngine(cfg.Projects[i].Name, e) e.SetTimerScheduler(timerSched) diff --git a/config/config.go b/config/config.go index 793aad7de8..250f97cae5 100644 --- a/config/config.go +++ b/config/config.go @@ -106,6 +106,7 @@ type Config struct { OutgoingRateLimit OutgoingRateLimitConfig `toml:"outgoing_rate_limit"` // outgoing message throttling Relay RelayConfig `toml:"relay"` // bot-to-bot relay behavior Cron CronConfig `toml:"cron"` + Timer TimerConfig `toml:"timer"` Queue QueueConfig `toml:"queue"` Webhook WebhookConfig `toml:"webhook"` Bridge BridgeConfig `toml:"bridge"` @@ -136,6 +137,16 @@ type CronConfig struct { SessionMode string `toml:"session_mode"` // default session mode: "" or "reuse" (default) or "new_per_run" } +// TimerConfig controls one-shot timer behavior. +type TimerConfig struct { + Enabled *bool `toml:"enabled"` // default true; set to false to disable /timer command + Silent *bool `toml:"silent"` // suppress timer fire notification; nil = inherit from [cron].silent + SessionMode string `toml:"session_mode"` // default session mode; "" = inherit from [cron].session_mode + MaxPendingJobs int `toml:"max_pending_jobs"` // max pending (unfired) timers; 0 = default (50) +} + +const DefaultTimerMaxPendingJobs = 50 + // QueueConfig controls the per-session message queue. type QueueConfig struct { MaxDepth *int `toml:"max_depth"` // max queued messages per session; default 5 @@ -909,6 +920,42 @@ func EffectiveDisplay(cfg *Config, proj *ProjectConfig) (mode string, thinkingMe return } +// allowedShellBases lists base names (lowercase, without .exe) that are +// accepted in the shell configuration. Any path whose base name matches +// one of these entries (with or without .exe suffix) is allowed. +var allowedShellBases = map[string]bool{ + "sh": true, + "bash": true, + "zsh": true, + "fish": true, + "dash": true, + "ksh": true, + "cmd": true, + "powershell": true, + "pwsh": true, +} + +// ValidateShellBinary checks whether the configured shell path refers to a +// recognized shell. Returns nil if empty (platform default will be used) or +// if the base name is on the allow list. Returns an error otherwise. +func ValidateShellBinary(shell string) error { + if shell == "" { + return nil + } + base := filepath.Base(shell) + // On non-Windows, filepath.Base doesn't handle backslashes, so also try + // splitting on backslash for Windows-style paths in config files. + if idx := strings.LastIndex(base, "\\"); idx >= 0 { + base = base[idx+1:] + } + base = strings.ToLower(base) + base = strings.TrimSuffix(base, ".exe") + if allowedShellBases[base] { + return nil + } + return fmt.Errorf("unsupported shell %q: allowed shells are sh, bash, zsh, fish, dash, ksh, cmd, powershell, pwsh", shell) +} + // EffectiveShell returns the shell binary, flag, and init command for the project. // Resolution: per-project > global > platform default. // The flag is auto-detected: "/C" for cmd, "-Command" for powershell/pwsh, "-c" for everything else. diff --git a/config/config_test.go b/config/config_test.go index 8a23a2aa3f..798617cb4e 100644 --- a/config/config_test.go +++ b/config/config_test.go @@ -3313,3 +3313,43 @@ func TestRemoveGlobalProvider_CleansUpProviderRefs(t *testing.T) { t.Errorf("proj2 provider_refs: want [], got %v", refs2) } } + +func TestValidateShellBinary(t *testing.T) { + tests := []struct { + input string + wantErr bool + }{ + {"", false}, + {"sh", false}, + {"bash", false}, + {"/bin/bash", false}, + {"/usr/local/bin/zsh", false}, + {"fish", false}, + {"dash", false}, + {"ksh", false}, + {"cmd", false}, + {"cmd.exe", false}, + {"powershell", false}, + {"powershell.exe", false}, + {"pwsh", false}, + {"pwsh.exe", false}, + {"C:\\Windows\\System32\\cmd.exe", false}, + {"rm", true}, + {"/bin/rm", true}, + {"python", true}, + {"node", true}, + {"/usr/bin/env", true}, + {"cat", true}, + } + for _, tt := range tests { + t.Run(tt.input, func(t *testing.T) { + err := ValidateShellBinary(tt.input) + if tt.wantErr && err == nil { + t.Errorf("ValidateShellBinary(%q) = nil, want error", tt.input) + } + if !tt.wantErr && err != nil { + t.Errorf("ValidateShellBinary(%q) = %v, want nil", tt.input, err) + } + }) + } +} diff --git a/core/engine.go b/core/engine.go index de94e2d962..3bcb33d850 100644 --- a/core/engine.go +++ b/core/engine.go @@ -1605,152 +1605,11 @@ func (e *Engine) executeTimerShell(p Platform, replyCtx any, job *TimerJob) erro if workDir == "" { workDir, _ = os.Getwd() } - timeout := job.ExecutionTimeout() if timeout <= 0 { timeout = 60 * time.Second } - - cmdLabel := truncateStr(job.Exec, 60) - - ctx, cancel := context.WithTimeout(e.ctx, timeout) - defer cancel() - - var shellCmd *exec.Cmd - if runtime.GOOS == "windows" { - shellCmd = exec.CommandContext(ctx, "powershell.exe", "-NoProfile", "-ExecutionPolicy", "Bypass", "-Command", job.Exec) - } else { - shellCmd = exec.CommandContext(ctx, "sh", "-c", job.Exec) - } - shellCmd.Dir = workDir - - stdout, err := shellCmd.StdoutPipe() - if err != nil { - return fmt.Errorf("shell: stdout pipe: %w", err) - } - stderr, err := shellCmd.StderrPipe() - if err != nil { - return fmt.Errorf("shell: stderr pipe: %w", err) - } - - if err := shellCmd.Start(); err != nil { - e.send(p, replyCtx, fmt.Sprintf("⏰ ❌ `%s`\nerror: failed to start: %v", cmdLabel, err)) - return fmt.Errorf("shell: start: %w", err) - } - - var mu sync.Mutex - var buf bytes.Buffer - doneCh := make(chan struct{}) - - readPipe := func(r io.Reader) { - scanner := bufio.NewScanner(r) - scanner.Buffer(make([]byte, 0, 64*1024), 64*1024) - for scanner.Scan() { - mu.Lock() - if buf.Len() > 0 { - buf.WriteByte('\n') - } - buf.WriteString(scanner.Text()) - mu.Unlock() - } - } - var pipeWg sync.WaitGroup - pipeWg.Add(2) - go func() { defer pipeWg.Done(); readPipe(stdout) }() - go func() { defer pipeWg.Done(); readPipe(stderr) }() - - go func() { - pipeWg.Wait() - _ = shellCmd.Wait() - close(doneCh) - }() - - select { - case <-doneCh: - return e.finishCronShell(p, replyCtx, shellCmd, &mu, &buf, cmdLabel) - case <-ctx.Done(): - killAndWait(shellCmd, doneCh) - mu.Lock() - output := buf.String() - mu.Unlock() - msg := fmt.Sprintf("⏰ ⚠️ timeout: `%s`", cmdLabel) - if output != "" { - msg = fmt.Sprintf("⏰ ⚠️ timeout: `%s`\n\n%s", cmdLabel, truncateStr(output, 3000)) - } - e.send(p, replyCtx, msg) - return fmt.Errorf("shell command timed out") - case <-time.After(quickFinishTimeout): - } - - // Long-running command — try in-place updates - var previewHandle any - var useUpdate bool - if _, ok := p.(MessageUpdater); ok { - if starter, ok := p.(PreviewStarter); ok { - mu.Lock() - output := buf.String() - mu.Unlock() - progressMsg := fmt.Sprintf("⏰ ⏳ `%s`", cmdLabel) - if output != "" { - progressMsg = fmt.Sprintf("⏰ ⏳ `%s`\n\n%s", cmdLabel, truncateStr(output, 3000)) - } - handle, err := starter.SendPreviewStart(e.ctx, replyCtx, progressMsg) - if err == nil && handle != nil { - previewHandle = handle - useUpdate = true - } - } - } - if !useUpdate { - e.send(p, replyCtx, fmt.Sprintf("⏰ ⏳ `%s`", cmdLabel)) - } - - updateDone := make(chan struct{}) - if useUpdate { - go func() { - ticker := time.NewTicker(2 * time.Second) - defer ticker.Stop() - for { - select { - case <-ticker.C: - mu.Lock() - output := buf.String() - mu.Unlock() - msg := fmt.Sprintf("⏰ ⏳ `%s`", cmdLabel) - if output != "" { - msg = fmt.Sprintf("⏰ ⏳ `%s`\n\n%s", cmdLabel, truncateStr(output, 3000)) - } - _ = updaterFor(p).UpdateMessage(e.ctx, previewHandle, msg) - case <-updateDone: - return - case <-ctx.Done(): - return - } - } - }() - } - - select { - case <-doneCh: - close(updateDone) - return e.finishCronShell(p, replyCtx, shellCmd, &mu, &buf, cmdLabel, useUpdate, previewHandle) - case <-ctx.Done(): - close(updateDone) - killAndWait(shellCmd, doneCh) - mu.Lock() - output := buf.String() - mu.Unlock() - msg := fmt.Sprintf("⏰ ⚠️ timeout: `%s`", cmdLabel) - if output != "" { - msg = fmt.Sprintf("⏰ ⚠️ timeout: `%s`\n\n%s", cmdLabel, truncateStr(output, 3000)) - } - if useUpdate { - _ = updaterFor(p).UpdateMessage(e.ctx, previewHandle, msg) - } else { - e.send(p, replyCtx, msg) - } - return fmt.Errorf("shell command timed out") - } + return e.executeScheduledShell(p, replyCtx, job.Exec, workDir, timeout) } func cronRunTitle(job *CronJob) string { @@ -1783,18 +1642,23 @@ func (e *Engine) executeCronShell(p Platform, replyCtx any, job *CronJob) error if workDir == "" { workDir, _ = os.Getwd() } - timeout := job.ExecutionTimeout() if timeout <= 0 { timeout = 60 * time.Second } + return e.executeScheduledShell(p, replyCtx, job.Exec, workDir, timeout) +} - cmdLabel := truncateStr(job.Exec, 60) +// executeScheduledShell is the shared implementation for both timer and cron +// shell execution. It handles command lifecycle, pipe reading, progress +// updates, and timeout handling. +func (e *Engine) executeScheduledShell(p Platform, replyCtx any, execStr string, workDir string, timeout time.Duration) error { + cmdLabel := truncateStr(execStr, 60) ctx, cancel := context.WithTimeout(e.ctx, timeout) defer cancel() - shellCmd := shellExecCommand(ctx, e.shell, e.shellFlag, e.shellProfile, job.Exec) + shellCmd := shellExecCommand(ctx, e.shell, e.shellFlag, e.shellProfile, execStr) shellCmd.Dir = workDir stdout, err := shellCmd.StdoutPipe() @@ -1827,10 +1691,8 @@ func (e *Engine) executeCronShell(p Platform, replyCtx any, job *CronJob) error mu.Unlock() } } - // Use a WaitGroup so both pipe-reader goroutines drain completely before - // doneCh is closed. Without this, shellCmd.Wait() can return (closing the - // pipe write-ends) while the scanners still have unread data in the OS - // buffer, causing finishCronShell to read a truncated output. + // WaitGroup ensures both pipe-reader goroutines drain completely before + // doneCh is closed, preventing truncated output. var pipeWg sync.WaitGroup pipeWg.Add(2) go func() { defer pipeWg.Done(); readPipe(stdout) }() @@ -1842,7 +1704,6 @@ func (e *Engine) executeCronShell(p Platform, replyCtx any, job *CronJob) error close(doneCh) }() - // Wait briefly to see if the command finishes quickly select { case <-doneCh: return e.finishCronShell(p, replyCtx, shellCmd, &mu, &buf, cmdLabel) @@ -1858,10 +1719,8 @@ func (e *Engine) executeCronShell(p Platform, replyCtx any, job *CronJob) error e.send(p, replyCtx, msg) return fmt.Errorf("shell command timed out") case <-time.After(quickFinishTimeout): - // Still running — fall through to progress mode } - // Long-running command. Try in-place updates. var previewHandle any var useUpdate bool if _, ok := p.(MessageUpdater); ok { diff --git a/core/engine_test.go b/core/engine_test.go index 95864e50a6..d9aaa57db6 100644 --- a/core/engine_test.go +++ b/core/engine_test.go @@ -15148,3 +15148,239 @@ func TestAgentSystemPrompt_DocumentsAudioVideoFlags(t *testing.T) { t.Error("AgentSystemPrompt missing the 'Do NOT downgrade' anti-regression line") } } + +// --- ExecuteTimerJob integration tests --- + +func TestExecuteTimerJob_MultiWorkspaceSessionKeyPrefix(t *testing.T) { + dir := t.TempDir() + store, err := NewTimerStore(dir) + if err != nil { + t.Fatalf("NewTimerStore() error = %v", err) + } + scheduler := NewTimerScheduler(store) + + platform := &stubCronReplyTargetPlatform{ + stubPlatformEngine: stubPlatformEngine{n: "slack"}, + } + agentSession := newResultAgentSession("timer done") + agent := &resultAgent{session: agentSession} + + e := NewEngine("test", agent, []Platform{platform}, "", LangEnglish) + defer e.cancel() + e.timerScheduler = scheduler + + prefixedKey := "/home/user/workspace:slack:C001:U001" + job := &TimerJob{ + ID: "timer-ws-1", + SessionKey: prefixedKey, + Prompt: "workspace timer", + Description: "WS timer", + ScheduledAt: time.Now().Add(-time.Second), + CreatedAt: time.Now(), + } + if err := store.Add(job); err != nil { + t.Fatalf("store.Add() error = %v", err) + } + + if err := e.ExecuteTimerJob(job); err != nil { + t.Fatalf("ExecuteTimerJob() with workspace-prefixed key error = %v", err) + } + + sent := platform.getSent() + if len(sent) < 1 { + t.Fatalf("expected at least one message sent, got %d", len(sent)) + } + + if job.SessionKey != prefixedKey { + t.Fatalf("job.SessionKey = %q, want unchanged %q", job.SessionKey, prefixedKey) + } +} + +func TestExecuteTimerJob_ShellExecFailsOnBadCommand(t *testing.T) { + dir := t.TempDir() + store, err := NewTimerStore(dir) + if err != nil { + t.Fatalf("NewTimerStore() error = %v", err) + } + scheduler := NewTimerScheduler(store) + + platform := &stubCronReplyTargetPlatform{ + stubPlatformEngine: stubPlatformEngine{n: "telegram"}, + } + agent := &resultAgent{session: newResultAgentSession("unused")} + + e := NewEngine("test", agent, []Platform{platform}, "", LangEnglish) + defer e.cancel() + e.timerScheduler = scheduler + e.SetShell("/nonexistent-shell-binary", "-c", "") + + job := &TimerJob{ + ID: "timer-bad-shell", + SessionKey: "telegram:chat1:user1", + Exec: "echo hello", + Description: "bad shell", + ScheduledAt: time.Now().Add(-time.Second), + CreatedAt: time.Now(), + } + if err := store.Add(job); err != nil { + t.Fatalf("store.Add() error = %v", err) + } + + err = e.ExecuteTimerJob(job) + if err == nil { + t.Fatalf("ExecuteTimerJob() expected error for bad shell, got nil") + } + if !strings.Contains(err.Error(), "start") { + t.Fatalf("ExecuteTimerJob() error = %q, want 'start' in error", err.Error()) + } + + sent := platform.getSent() + foundError := false + for _, msg := range sent { + if strings.Contains(msg, "❌") || strings.Contains(msg, "failed to start") { + foundError = true + break + } + } + if !foundError { + t.Fatalf("expected error message in sent, got: %v", sent) + } +} + +func TestExecuteTimerJob_ExpandsSlashSkill(t *testing.T) { + skillRoot := t.TempDir() + writeSkillFile(t, filepath.Join(skillRoot, "check-status", "SKILL.md"), "Status check skill") + + dir := t.TempDir() + store, err := NewTimerStore(dir) + if err != nil { + t.Fatalf("NewTimerStore() error = %v", err) + } + scheduler := NewTimerScheduler(store) + + platform := &stubCronReplyTargetPlatform{ + stubPlatformEngine: stubPlatformEngine{n: "discord"}, + } + agentSession := newResultAgentSession("skill result") + agent := &resultAgent{session: agentSession} + + e := NewEngine("test", agent, []Platform{platform}, "", LangEnglish) + defer e.cancel() + e.timerScheduler = scheduler + e.skills.SetDirs([]string{skillRoot}) + + job := &TimerJob{ + ID: "timer-skill-1", + SessionKey: "discord:channel-1:user-1", + Prompt: "/check-status weekly", + Description: "Weekly check", + ScheduledAt: time.Now().Add(-time.Second), + CreatedAt: time.Now(), + } + if err := store.Add(job); err != nil { + t.Fatalf("store.Add() error = %v", err) + } + + if err := e.ExecuteTimerJob(job); err != nil { + t.Fatalf("ExecuteTimerJob() error = %v", err) + } + + if len(agentSession.sentPrompts) != 1 { + t.Fatalf("sentPrompts = %d, want 1: %#v", len(agentSession.sentPrompts), agentSession.sentPrompts) + } + got := agentSession.sentPrompts[0] + if !strings.Contains(got, "## Skill") { + t.Errorf("agent prompt should contain skill expansion, got: %s", got) + } + if !strings.Contains(got, "check-status") { + t.Errorf("agent prompt should contain skill name, got: %s", got) + } + if !strings.Contains(got, "weekly") { + t.Errorf("agent prompt should contain args 'weekly', got: %s", got) + } +} + +func TestExecuteTimerJob_ShellUsesConfiguredShell(t *testing.T) { + dir := t.TempDir() + store, err := NewTimerStore(dir) + if err != nil { + t.Fatalf("NewTimerStore() error = %v", err) + } + scheduler := NewTimerScheduler(store) + + platform := &stubCronReplyTargetPlatform{ + stubPlatformEngine: stubPlatformEngine{n: "telegram"}, + } + agent := &resultAgent{session: newResultAgentSession("unused")} + + e := NewEngine("test", agent, []Platform{platform}, "", LangEnglish) + defer e.cancel() + e.timerScheduler = scheduler + e.SetShell("/bin/sh", "-c", "") + + job := &TimerJob{ + ID: "timer-shell-cfg", + SessionKey: "telegram:chat1:user1", + Exec: "echo hello-timer", + Description: "shell test", + ScheduledAt: time.Now().Add(-time.Second), + CreatedAt: time.Now(), + } + if err := store.Add(job); err != nil { + t.Fatalf("store.Add() error = %v", err) + } + + if err := e.ExecuteTimerJob(job); err != nil { + t.Fatalf("ExecuteTimerJob() error = %v", err) + } + + sent := platform.getSent() + foundOutput := false + for _, msg := range sent { + if strings.Contains(msg, "hello-timer") { + foundOutput = true + break + } + } + if !foundOutput { + t.Fatalf("expected 'hello-timer' in output messages, got: %v", sent) + } +} + +func TestTimerScheduler_MaxPendingJobsEnforced(t *testing.T) { + dir := t.TempDir() + store, err := NewTimerStore(dir) + if err != nil { + t.Fatalf("NewTimerStore() error = %v", err) + } + scheduler := NewTimerScheduler(store) + scheduler.SetMaxPendingJobs(3) + + for i := 0; i < 3; i++ { + job := &TimerJob{ + ID: fmt.Sprintf("job-%d", i), + SessionKey: "telegram:chat:user", + Prompt: "test", + ScheduledAt: time.Now().Add(time.Hour), + CreatedAt: time.Now(), + } + if err := scheduler.AddJob(job); err != nil { + t.Fatalf("AddJob(%d) unexpected error: %v", i, err) + } + } + + overflowJob := &TimerJob{ + ID: "job-overflow", + SessionKey: "telegram:chat:user", + Prompt: "overflow", + ScheduledAt: time.Now().Add(time.Hour), + CreatedAt: time.Now(), + } + err = scheduler.AddJob(overflowJob) + if err == nil { + t.Fatalf("AddJob(overflow) expected error, got nil") + } + if !strings.Contains(err.Error(), "timer limit reached") { + t.Fatalf("AddJob(overflow) error = %q, want 'timer limit reached'", err.Error()) + } +} diff --git a/core/timer.go b/core/timer.go index bbd23dde58..8f86f72549 100644 --- a/core/timer.go +++ b/core/timer.go @@ -244,6 +244,7 @@ type TimerScheduler struct { timers map[string]*time.Timer // job ID → active timer defaultSilent bool defaultSessionMode string + maxPendingJobs int // 0 = use default (50) } // missedJobGracePeriod is how long after a missed fire time we still execute. @@ -276,6 +277,19 @@ func (ts *TimerScheduler) SetDefaultSessionMode(mode string) { ts.defaultSessionMode = NormalizeCronSessionMode(mode) } +// SetMaxPendingJobs sets the maximum number of pending (unfired) timer jobs allowed. +// Must be called before Start (not safe for concurrent use). +func (ts *TimerScheduler) SetMaxPendingJobs(max int) { + ts.maxPendingJobs = max +} + +func (ts *TimerScheduler) maxPending() int { + if ts.maxPendingJobs > 0 { + return ts.maxPendingJobs + } + return 50 +} + // IsSilent returns whether the timer job should suppress the start notification. func (ts *TimerScheduler) IsSilent(job *TimerJob) bool { if job.Silent != nil { @@ -335,6 +349,9 @@ func (ts *TimerScheduler) AddJob(job *TimerJob) error { if err := validateTimerJob(job); err != nil { return err } + if pending := ts.store.ListPending(); len(pending) >= ts.maxPending() { + return fmt.Errorf("timer limit reached: %d pending jobs (max %d)", len(pending), ts.maxPending()) + } job.SessionMode = NormalizeCronSessionMode(job.SessionMode) if err := ts.store.Add(job); err != nil { return err diff --git a/core/tts.go b/core/tts.go index 94eed521e9..8ee77770d5 100644 --- a/core/tts.go +++ b/core/tts.go @@ -12,6 +12,7 @@ import ( "net/http" "os" "os/exec" + "runtime" "strings" "sync" "time" @@ -187,10 +188,14 @@ func (q *QwenTTS) Synthesize(ctx context.Context, text string, opts TTSSynthesis } defer audioResp.Body.Close() - wavData, err := io.ReadAll(audioResp.Body) + const maxTTSAudioSize = 20 * 1024 * 1024 // 20 MB + wavData, err := io.ReadAll(io.LimitReader(audioResp.Body, maxTTSAudioSize+1)) if err != nil { return nil, "", fmt.Errorf("qwen tts: read audio: %w", err) } + if len(wavData) > maxTTSAudioSize { + return nil, "", fmt.Errorf("qwen tts: audio response exceeds %d bytes", maxTTSAudioSize) + } return wavData, "wav", nil } @@ -555,34 +560,59 @@ func (e *EspeakTTS) Synthesize(ctx context.Context, text string, opts TTSSynthes voice = e.Voice } - // Build espeak command + if runtime.GOOS == "windows" { + return e.synthesizeViaTempFile(ctx, text, voice, opts.Speed) + } + args := []string{ "-v", voice, - "-w", "/dev/stdout", // write WAV to stdout (Unix-only; not supported on Windows) + "-w", "/dev/stdout", } - - // Add speed option if specified if opts.Speed > 0 { - // espeak speed is in words per minute, default 160 - // Convert speed multiplier (0.5-2.0) to wpm wpm := int(160 * opts.Speed) args = append(args, "-s", fmt.Sprintf("%d", wpm)) } - - // Add text as argument args = append(args, text) - // Execute espeak command - // Use Output() instead of CombinedOutput() to avoid mixing stderr warnings with audio data cmd := exec.CommandContext(ctx, e.Path, args...) output, err := cmd.Output() if err != nil { return nil, "", fmt.Errorf("espeak: voice=%s text=%q: %w", voice, text, err) } - return output, "wav", nil } +func (e *EspeakTTS) synthesizeViaTempFile(ctx context.Context, text, voice string, speed float64) ([]byte, string, error) { + tmpFile, err := os.CreateTemp("", "espeak_tts_*.wav") + if err != nil { + return nil, "", fmt.Errorf("espeak: create temp file: %w", err) + } + tmpPath := tmpFile.Name() + _ = tmpFile.Close() + defer func() { _ = os.Remove(tmpPath) }() + + args := []string{"-v", voice, "-w", tmpPath} + if speed > 0 { + wpm := int(160 * speed) + args = append(args, "-s", fmt.Sprintf("%d", wpm)) + } + args = append(args, text) + + cmd := exec.CommandContext(ctx, e.Path, args...) + if output, err := cmd.CombinedOutput(); err != nil { + return nil, "", fmt.Errorf("espeak: voice=%s text=%q: %w, output: %s", voice, text, err, string(output)) + } + + audioData, err := os.ReadFile(tmpPath) + if err != nil { + return nil, "", fmt.Errorf("espeak: read output file: %w", err) + } + if len(audioData) == 0 { + return nil, "", fmt.Errorf("espeak: produced empty audio file") + } + return audioData, "wav", nil +} + // ────────────────────────────────────────────────────────────── // PicoTTS — Google Pico TTS (better quality than espeak, offline) // ────────────────────────────────────────────────────────────── diff --git a/platform/qq/qq.go b/platform/qq/qq.go index 294c09bc31..6471f355de 100644 --- a/platform/qq/qq.go +++ b/platform/qq/qq.go @@ -703,6 +703,8 @@ func stripCQCodes(s string) string { return result.String() } +const maxDownloadSize = 100 * 1024 * 1024 // 100 MB + func downloadLargeFile(url string) ([]byte, string, error) { client := &http.Client{Timeout: 120 * time.Second} resp, err := client.Get(url) @@ -715,10 +717,13 @@ func downloadLargeFile(url string) ([]byte, string, error) { return nil, "", fmt.Errorf("HTTP %d", resp.StatusCode) } - data, err := io.ReadAll(resp.Body) + data, err := io.ReadAll(io.LimitReader(resp.Body, maxDownloadSize+1)) if err != nil { return nil, "", err } + if len(data) > maxDownloadSize { + return nil, "", fmt.Errorf("file exceeds max download size (%d bytes)", maxDownloadSize) + } mime := resp.Header.Get("Content-Type") if mime == "" { @@ -727,6 +732,8 @@ func downloadLargeFile(url string) ([]byte, string, error) { return data, mime, nil } +const maxMediaDownloadSize = 20 * 1024 * 1024 // 20 MB for images/audio + func downloadFile(url string) ([]byte, string, error) { client := &http.Client{Timeout: 30 * time.Second} resp, err := client.Get(url) @@ -735,10 +742,13 @@ func downloadFile(url string) ([]byte, string, error) { } defer resp.Body.Close() - data, err := io.ReadAll(resp.Body) + data, err := io.ReadAll(io.LimitReader(resp.Body, maxMediaDownloadSize+1)) if err != nil { return nil, "", err } + if len(data) > maxMediaDownloadSize { + return nil, "", fmt.Errorf("media exceeds max download size (%d bytes)", maxMediaDownloadSize) + } mime := resp.Header.Get("Content-Type") if mime == "" {