Enhance scheduler tool execution by deduplicating identical calls and improve provider reinitialization for better runtime configuration handling

cmd/godex/scheduler.go

@@ -23,10 +23,14 @@ func (s *schedulerProviderGetter) GetProvider(cfg interface{}) (interface{}, err
 	providerName, _ := cfgMap["name"].(string)
 	providerModel, _ := cfgMap["model"].(string)
 
-	if (providerType == "unknown" || providerType == "current") && s.provider != nil {
-		providerType = s.provider.Type
-		providerName = s.provider.Name
-		providerModel = s.provider.Model
+	// The scheduler often requests a "current" provider. For providers like OpenRouter,
+	// the API key lives on the full config.Provider struct, so we must pass it through.
+	useCurrent := s.provider != nil && (providerType == "" || providerType == "unknown" || providerType == "current" || providerName == "" || providerName == "current")
+	if useCurrent {
+		// Reinitialize to ensure runtime auth (OpenRouter keys, env) is present for scheduler runs.
+		// NOTE: We intentionally do not override the model here to avoid mutating global config.
+		_ = providerModel
+		return agent.ReinitializeProvider(s.provider)
 	}
 
 	return agent.GetProviderFromConfig(&agent.ProviderConfig{
@@ -159,6 +163,11 @@ func handleSchedule(input string) {
 			return
 		}
 		if st.LastOutput == "" {
+
+			if st.LastError != "" {
+				fmt.Printf("Error for task %s:\n%s\n", id, st.LastError)
+				return
+			}
 			fmt.Printf("No output for task %s\n", id)
 			return
 		}

internal/agent/client.go

@@ -3,7 +3,6 @@ package agent
 import (
 	"context"
 	"fmt"
-	"log"
 	"strings"
 	"sync"
 
@@ -17,6 +16,8 @@ var (
 	providerCacheMu sync.Mutex
 	mcpExecutors    = map[string]*mcp.MCPToolExecutor{}
 	mcpExecutorsMu  sync.Mutex
+	toolsCache      = map[string][]providers.Tool{}
+	toolsCacheMu    sync.Mutex
 )
 
 type SendResult struct {
@@ -75,6 +76,12 @@ func SetProviderTools(provider *config.Provider, tools []providers.Tool) {
 	if err != nil {
 		return
 	}
+	if provider != nil {
+		key := cacheKey(provider)
+		toolsCacheMu.Lock()
+		toolsCache[key] = tools
+		toolsCacheMu.Unlock()
+	}
 	if pt, ok := p.(interface{ SetTools([]providers.Tool) }); ok {
 		pt.SetTools(tools)
 	}
@@ -107,7 +114,7 @@ func CloseProvider(cfg *config.Provider) {
 	}
 	providerCacheMu.Unlock()
 	if ok {
-		log.Printf("[agent] Closing provider for %s/%s", cfg.Type, cfg.Name)
+		//log.Printf("[agent] Closing provider for %s/%s", cfg.Type, cfg.Name)
 		_ = p.Close()
 	}
 
@@ -137,10 +144,29 @@ func GetProvider(cfg *config.Provider) (providers.Provider, error) {
 	if err != nil {
 		return nil, err
 	}
+	toolsCacheMu.Lock()
+	tools := toolsCache[key]
+	toolsCacheMu.Unlock()
+	if len(tools) > 0 {
+		if pt, ok := p.(interface{ SetTools([]providers.Tool) }); ok {
+			pt.SetTools(tools)
+		}
+	}
 	providerCache[key] = p
 	return p, nil
 }
 
+// ReinitializeProvider closes any cached provider instance for cfg and creates a fresh one.
+// This is useful when background workers (like the scheduler) need a clean client instance
+// with all settings (API keys, endpoint, params) preserved.
+func ReinitializeProvider(cfg *config.Provider) (providers.Provider, error) {
+	if cfg == nil {
+		return nil, fmt.Errorf("nil provider")
+	}
+	CloseProvider(cfg)
+	return GetProvider(cfg)
+}
+
 func cacheKey(cfg *config.Provider) string {
 	return fmt.Sprintf("%s|%s|%s|%s", cfg.Type, cfg.Name, cfg.Model, cfg.Endpoint)
 }

internal/providers/openrouter.go

@@ -89,6 +89,19 @@ type openRouterProvider struct {
 	statusCh         chan<- string
 }
 
+func (p *openRouterProvider) emitStatus(msg string) {
+	p.mu.Lock()
+	ch := p.statusCh
+	p.mu.Unlock()
+	if ch == nil {
+		return
+	}
+	select {
+	case ch <- msg:
+	default:
+	}
+}
+
 func init() {
 	Register("openrouter", newOpenRouterProvider)
 }
@@ -219,13 +232,13 @@ func (p *openRouterProvider) Send(ctx context.Context, prompt string) (string, e
 	var lastErr error
 	for attempt := 0; attempt <= maxRetries; attempt++ {
 		if attempt > 0 {
-			fmt.Printf("\n[OPENROUTER] Waiting %v before retry (attempt %d/%d)...\n", retryDelay, attempt, maxRetries)
+			p.emitStatus(fmt.Sprintf("OpenRouter: waiting %v before retry (%d/%d)", retryDelay, attempt, maxRetries))
 			select {
 			case <-ctx.Done():
 				return "", ctx.Err()
 			case <-time.After(retryDelay):
 			}
-			fmt.Printf("\n[OPENROUTER] Retrying request (attempt %d/%d)...\n", attempt, maxRetries)
+			p.emitStatus(fmt.Sprintf("OpenRouter: retrying request (%d/%d)", attempt, maxRetries))
 		}
 
 		req, err := http.NewRequestWithContext(ctx, "POST", p.baseURL+"/responses", bytes.NewReader(body))
@@ -242,7 +255,7 @@ func (p *openRouterProvider) Send(ctx context.Context, prompt string) (string, e
 		if err != nil {
 			lastErr = fmt.Errorf("request failed: %w", err)
 			if attempt < maxRetries {
-				fmt.Printf("[OPENROUTER00] retryable error: %v\n", lastErr)
+				p.emitStatus(fmt.Sprintf("OpenRouter: retryable error: %v", lastErr))
 			}
 			continue
 		}
@@ -253,7 +266,7 @@ func (p *openRouterProvider) Send(ctx context.Context, prompt string) (string, e
 		if resp.StatusCode == http.StatusTooManyRequests {
 			lastErr = fmt.Errorf("API error (status %d): %s", resp.StatusCode, string(bodyBytes))
 			if attempt < maxRetries {
-				fmt.Printf("[OPENROUTER] retryable error: %v\n", lastErr)
+				p.emitStatus(fmt.Sprintf("OpenRouter: retryable error: %v", lastErr))
 			}
 			continue
 		}

internal/scheduler/scheduler.go

@@ -559,23 +559,30 @@ func (s *Scheduler) executeWithTools(ctx context.Context, task *ScheduledTask) (
 			return finalResp, nil
 		}
 
-		prevNoTool = false
+			prevNoTool = false
 
-		var toolResults []string
-		for _, tc := range toolCalls {
-			toolName, ok := tc["name"].(string)
-			if !ok {
-				continue
-			}
-			args, ok := tc["arguments"].(map[string]interface{})
-			if !ok {
-				continue
-			}
+			var toolResults []string
+			seenThisRound := make(map[string]bool)
+			for _, tc := range toolCalls {
+				toolName, ok := tc["name"].(string)
+				if !ok {
+					continue
+				}
+				args, ok := tc["arguments"].(map[string]interface{})
+				if !ok {
+					continue
+				}
+				argsJSON, _ := json.Marshal(args)
+				sig := fmt.Sprintf("%s:%s", toolName, string(argsJSON))
+				if seenThisRound[sig] {
+					continue
+				}
+				seenThisRound[sig] = true
 
-			result, err := s.callTool(ctx, toolName, args, toolTimeout)
-			if err != nil {
-				toolResults = append(toolResults, fmt.Sprintf("ERROR: %v", err))
-			} else {
+				result, err := s.callTool(ctx, toolName, args, toolTimeout)
+				if err != nil {
+					toolResults = append(toolResults, fmt.Sprintf("ERROR: %v", err))
+				} else {
 				toolResults = append(toolResults, s.truncate(result, 2500))
 			}
 		}
@@ -584,13 +591,16 @@ func (s *Scheduler) executeWithTools(ctx context.Context, task *ScheduledTask) (
 			return finalResp, nil
 		}
 
-		currentToolCalls := make(map[string]bool)
-		for _, tc := range toolCalls {
-			name := tc["name"].(string)
-			args := tc["arguments"]
-			argsJson, _ := json.Marshal(args)
-			currentToolCalls[fmt.Sprintf("%s:%s", name, string(argsJson))] = true
-		}
+			currentToolCalls := make(map[string]bool)
+			for _, tc := range toolCalls {
+				name, ok := tc["name"].(string)
+				if !ok {
+					continue
+				}
+				args := tc["arguments"]
+				argsJSON, _ := json.Marshal(args)
+				currentToolCalls[fmt.Sprintf("%s:%s", name, string(argsJSON))] = true
+			}
 
 		hasRepeatedCalls := false
 		for sig := range currentToolCalls {

internal/scheduler/scheduler_dedupe_test.go

@@ -0,0 +1,81 @@
+package scheduler
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/cheikh2shift/godex/internal/mcp"
+)
+
+type fakeProviderGetter struct {
+	p interface{}
+}
+
+func (f *fakeProviderGetter) GetProvider(cfg interface{}) (interface{}, error) {
+	return f.p, nil
+}
+
+type fakeLLMProvider struct {
+	resp string
+}
+
+func (f *fakeLLMProvider) Send(ctx context.Context, prompt string) (string, error) {
+	return f.resp, nil
+}
+func (f *fakeLLMProvider) SetThinkCallback(cb func(string)) {}
+func (f *fakeLLMProvider) SupportsNativeToolCalls() bool     { return false }
+
+type countingToolServer struct {
+	name  string
+	tools []mcp.Tool
+	calls int
+}
+
+func (c *countingToolServer) Name() string       { return c.name }
+func (c *countingToolServer) Tools() []mcp.Tool  { return c.tools }
+func (c *countingToolServer) CallTool(ctx context.Context, name string, args map[string]interface{}) (string, error) {
+	c.calls++
+	return "ok", nil
+}
+
+func TestExecuteWithTools_DedupesIdenticalToolCallsInOneResponse(t *testing.T) {
+	s := &Scheduler{}
+	server := &countingToolServer{
+		name: "scheduler",
+		tools: []mcp.Tool{
+			{Name: "scheduler", Description: "schedule"},
+		},
+	}
+	s.servers = []ToolServer{server}
+	s.provider = &fakeProviderGetter{p: &fakeLLMProvider{
+		// Two identical tool calls in one response. Should execute only once.
+		resp: "```json\n" +
+			"{\"name\":\"scheduler\",\"arguments\":{\"prompt\":\"x\",\"interval_sec\":60}}\n" +
+			"```\n" +
+			"```json\n" +
+			"{\"name\":\"scheduler\",\"arguments\":{\"prompt\":\"x\",\"interval_sec\":60}}\n" +
+			"```",
+	}}
+	s.maxRounds = 1
+	s.toolTimeout = 1
+
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+	defer cancel()
+
+	task := &ScheduledTask{
+		Prompt:        "ignored",
+		ProviderType:  "current",
+		ProviderName:  "current",
+		ProviderModel: "current",
+	}
+
+	_, err := s.executeWithTools(ctx, task)
+	if err != nil {
+		t.Fatalf("executeWithTools: %v", err)
+	}
+	if server.calls != 1 {
+		t.Fatalf("expected 1 tool execution, got %d", server.calls)
+	}
+}
+