From 89facc03ee0b17894e892843fa94463c42cd3354 Mon Sep 17 00:00:00 2001 From: Cytown Date: Sun, 8 Mar 2026 01:47:55 +0800 Subject: [PATCH] fix according to code review --- cmd/picoclaw/internal/gateway/helpers.go | 38 ++++--- pkg/agent/loop.go | 121 ++++++++++++++++++++++- pkg/logger/logger.go | 2 +- pkg/logger/logger_test.go | 4 + 4 files changed, 148 insertions(+), 17 deletions(-) diff --git a/cmd/picoclaw/internal/gateway/helpers.go b/cmd/picoclaw/internal/gateway/helpers.go index fecb8cc88..663697d74 100644 --- a/cmd/picoclaw/internal/gateway/helpers.go +++ b/cmd/picoclaw/internal/gateway/helpers.go @@ -233,11 +233,8 @@ func gatewayCmd(debug bool) error { } logger.Infof(" New model is '%s', recreating provider...", newModel) - if cp, ok := provider.(providers.StatefulProvider); ok { - cp.Close() - } - // Create new provider from updated config + // Create new provider from updated config first to ensure validity // This will use the correct API key and settings from newCfg.ModelList newProvider, newModelID, err := providers.CreateProvider(newCfg) if err != nil { @@ -246,20 +243,37 @@ func gatewayCmd(debug bool) error { continue } - provider = newProvider if newModelID != "" { newCfg.Agents.Defaults.ModelName = newModelID } + // Use the atomic reload method on AgentLoop to safely swap provider and config. + // This handles locking internally to prevent races with in-flight LLM calls + // and concurrent reads of registry/config while the swap occurs. + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + if err := agentLoop.ReloadProviderAndConfig(ctx, newProvider, newCfg); err != nil { + logger.Errorf(" ⚠ Error reloading agent loop: %v", err) + // Close the newly created provider since it wasn't adopted + if cp, ok := newProvider.(providers.StatefulProvider); ok { + cp.Close() + } + logger.Warn(" Continuing with old provider and config") + continue + } + + // Update local references only after successful atomic reload + if cp, ok := provider.(providers.StatefulProvider); ok { + cp.Close() + } + provider = newProvider + // Update agent loop provider and models - agentLoop.SetProvider(provider, newCfg) + //agentLoop.SetProvider(provider, newCfg) - logger.Info(" ✓ Provider and agents updated successfully") - - // Update the config reference for other operations - // Note: Some changes (like channel configs) may require restart to take full effect cfg = newCfg - logger.Info(" ✓ Configuration reloaded successfully") + logger.Info(" ✓ Provider and configuration reloaded successfully (thread-safe)") } } } @@ -291,7 +305,7 @@ func setupConfigWatcherPolling(configPath string, debug bool) (chan *config.Conf // Check if file changed (modification time or size changed) if currentModTime.After(lastModTime) || currentSize != lastSize { if debug { - logger.DebugSF("🔍 Config file change detected") + logger.Debugf("🔍 Config file change detected") } // Debounce - wait a bit to ensure file write is complete diff --git a/pkg/agent/loop.go b/pkg/agent/loop.go index 7153d7146..bfbd1732d 100644 --- a/pkg/agent/loop.go +++ b/pkg/agent/loop.go @@ -46,6 +46,7 @@ type AgentLoop struct { channelManager *channels.Manager mediaStore media.MediaStore transcriber voice.Transcriber + mu sync.RWMutex } // processOptions configures how a message is processed @@ -368,12 +369,110 @@ func (al *AgentLoop) SetChannelManager(cm *channels.Manager) { al.channelManager = cm } -// SetProvider updates the LLM provider for all agents in the registry -// and updates their model configurations. -func (al *AgentLoop) SetProvider(provider providers.LLMProvider, cfg *config.Config) { +// ReloadProviderAndConfig atomically swaps the provider and config with proper synchronization. +// It uses a context to allow timeout control from the caller. +// Returns an error if the reload fails or context is cancelled. +func (al *AgentLoop) ReloadProviderAndConfig(ctx context.Context, provider providers.LLMProvider, cfg *config.Config) error { + // Validate inputs + if provider == nil { + return fmt.Errorf("provider cannot be nil") + } + if cfg == nil { + return fmt.Errorf("config cannot be nil") + } + + // Check context before starting + if err := ctx.Err(); err != nil { + return fmt.Errorf("context cancelled before reload: %w", err) + } + + // Create new registry with updated config and provider + // Wrap in defer/recover to handle any panics gracefully + var registry *AgentRegistry + done := make(chan struct{}, 1) + + go func() { + defer func() { + if r := recover(); r != nil { + logger.ErrorCF("agent", "Panic during registry creation", + map[string]any{"panic": r}) + } + close(done) + }() + + registry = NewAgentRegistry(cfg, provider) + }() + + // Wait for completion or context cancellation + select { + case <-done: + if registry == nil { + return fmt.Errorf("registry creation failed (nil result)") + } + case <-ctx.Done(): + return fmt.Errorf("context cancelled during registry creation: %w", ctx.Err()) + } + + // Check context again before proceeding + if err := ctx.Err(); err != nil { + return fmt.Errorf("context cancelled after registry creation: %w", err) + } + + // Ensure shared tools are re-registered on the new registry + registerSharedTools(cfg, al.bus, registry, provider) + + // Atomically swap the config and registry under write lock + // This ensures readers see a consistent pair + al.mu.Lock() + oldRegistry := al.registry + + // Store new values al.cfg = cfg - registry := NewAgentRegistry(cfg, provider) al.registry = registry + + // Also update fallback chain with new config + al.fallback = providers.NewFallbackChain(providers.NewCooldownTracker()) + + al.mu.Unlock() + + // Close old provider after releasing the lock + // This prevents blocking readers while closing + if oldProvider, ok := extractProvider(oldRegistry); ok { + if stateful, ok := oldProvider.(providers.StatefulProvider); ok { + // Give in-flight requests a moment to complete + // Use a reasonable timeout that balances cleanup vs resource usage + select { + case <-time.After(100 * time.Millisecond): + stateful.Close() + case <-ctx.Done(): + // Context cancelled, close immediately but log warning + logger.WarnCF("agent", "Context cancelled during provider cleanup, forcing close", + map[string]any{"error": ctx.Err()}) + stateful.Close() + } + } + } + + logger.InfoCF("agent", "Provider and config reloaded successfully", + map[string]any{ + "model": cfg.Agents.Defaults.GetModelName(), + }) + + return nil +} + +// GetRegistry returns the current registry (thread-safe) +func (al *AgentLoop) GetRegistry() *AgentRegistry { + al.mu.RLock() + defer al.mu.RUnlock() + return al.registry +} + +// GetConfig returns the current config (thread-safe) +func (al *AgentLoop) GetConfig() *config.Config { + al.mu.RLock() + defer al.mu.RUnlock() + return al.cfg } // SetMediaStore injects a MediaStore for media lifecycle management. @@ -991,6 +1090,7 @@ func (al *AgentLoop) runLLMIteration( map[string]any{ "agent_id": agent.ID, "iteration": iteration, + "model": activeModel, "error": err.Error(), }) return "", iteration, fmt.Errorf("LLM call failed after retries: %w", err) @@ -1627,3 +1727,16 @@ func extractParentPeer(msg bus.InboundMessage) *routing.RoutePeer { } return &routing.RoutePeer{Kind: parentKind, ID: parentID} } + +// Helper to extract provider from registry for cleanup +func extractProvider(registry *AgentRegistry) (providers.LLMProvider, bool) { + if registry == nil { + return nil, false + } + // Get any agent to access the provider + defaultAgent := registry.GetDefaultAgent() + if defaultAgent == nil { + return nil, false + } + return defaultAgent.Provider, true +} diff --git a/pkg/logger/logger.go b/pkg/logger/logger.go index 08c0ea6ff..f1313fe89 100644 --- a/pkg/logger/logger.go +++ b/pkg/logger/logger.go @@ -168,7 +168,7 @@ func DebugC(component string, message string) { logMessage(DEBUG, component, message, nil) } -func DebugSF(message string, ss ...any) { +func Debugf(message string, ss ...any) { logMessage(DEBUG, "", fmt.Sprintf(message, ss...), nil) } diff --git a/pkg/logger/logger_test.go b/pkg/logger/logger_test.go index 6e6f8dfa8..8170a618b 100644 --- a/pkg/logger/logger_test.go +++ b/pkg/logger/logger_test.go @@ -123,17 +123,21 @@ func TestLoggerHelperFunctions(t *testing.T) { SetLevel(INFO) Debug("This should not log") + Debugf("this should not log") Info("This should log") Warn("This should log") Error("This should log") InfoC("test", "Component message") InfoF("Fields message", map[string]any{"key": "value"}) + Infof("test from %v", "Infof") WarnC("test", "Warning with component") ErrorF("Error with fields", map[string]any{"error": "test"}) + Errorf("test from %v", "Errorf") SetLevel(DEBUG) DebugC("test", "Debug with component") + Debugf("test from %v", "Debugf") WarnF("Warning with fields", map[string]any{"key": "value"}) }