fix(core): use configurable shell for timer exec + add limits and validation

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

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)

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.

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

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 {