From c24037820038dc1d6ba70e8288dabf04305dfb0c Mon Sep 17 00:00:00 2001 From: avaksru Date: Thu, 19 Mar 2026 10:28:05 +0300 Subject: [PATCH] gci-fmt --- go.mod | 7 - pkg/channels/manager.go | 53 ++++- pkg/channels/mqtt/init.go | 9 +- pkg/channels/mqtt/mqtt.go | 24 +- pkg/config/config.go | 56 ++--- pkg/gateway/gateway.go | 451 ++++++++++++++++++++++++++++++++++++-- 6 files changed, 525 insertions(+), 75 deletions(-) diff --git a/go.mod b/go.mod index 2b99d9fc1..65c561d91 100644 --- a/go.mod +++ b/go.mod @@ -94,15 +94,8 @@ require ( github.com/valyala/fastjson v1.6.10 // indirect github.com/yosida95/uritemplate/v3 v3.0.2 // indirect golang.org/x/arch v0.24.0 // indirect -<<<<<<< HEAD golang.org/x/crypto v0.49.0 golang.org/x/net v0.52.0 golang.org/x/sync v0.20.0 // indirect golang.org/x/sys v0.42.0 // indirect -======= - golang.org/x/crypto v0.48.0 - golang.org/x/net v0.51.0 - golang.org/x/sync v0.19.0 // indirect - golang.org/x/sys v0.41.0 // indirect ->>>>>>> 1038a05 (channel mqtt) ) diff --git a/pkg/channels/manager.go b/pkg/channels/manager.go index d88f20aa3..58985ebd7 100644 --- a/pkg/channels/manager.go +++ b/pkg/channels/manager.go @@ -158,8 +158,13 @@ func (m *Manager) RecordReactionUndo(channel, chatID string, undo func()) { } // preSend handles typing stop, reaction undo, and placeholder editing before sending a message. -// Returns true if the message was already delivered (skip Send). -func (m *Manager) preSend(ctx context.Context, name string, msg bus.OutboundMessage, ch Channel) bool { +// Returns true if the message was edited into a placeholder (skip Send). +func (m *Manager) preSend( + ctx context.Context, + name string, + msg bus.OutboundMessage, + ch Channel, +) bool { key := name + ":" + msg.ChatID // 1. Stop typing @@ -206,7 +211,11 @@ func (m *Manager) preSend(ctx context.Context, name string, msg bus.OutboundMess return false } -func NewManager(cfg *config.Config, messageBus *bus.MessageBus, store media.MediaStore) (*Manager, error) { +func NewManager( + cfg *config.Config, + messageBus *bus.MessageBus, + store media.MediaStore, +) (*Manager, error) { m := &Manager{ channels: make(map[string]Channel), workers: make(map[string]*channelWorker), @@ -401,7 +410,9 @@ func (m *Manager) initChannels(channels *config.ChannelsConfig) error { m.initChannel("irc", "IRC") } - if m.config.Channels.MQTT.Enabled && m.config.Channels.MQTT.Broker != "" && m.config.Channels.MQTT.ClientID != "" && len(m.config.Channels.MQTT.SubscribeTopics) > 0 { + if m.config.Channels.MQTT.Enabled && m.config.Channels.MQTT.Broker != "" && + m.config.Channels.MQTT.ClientID != "" && + len(m.config.Channels.MQTT.SubscribeTopics) > 0 { m.initChannel("mqtt", "MQTT") } @@ -623,7 +634,12 @@ func (m *Manager) runWorker(ctx context.Context, name string, w *channelWorker) // - ErrNotRunning / ErrSendFailed: permanent, no retry // - ErrRateLimit: fixed delay retry // - ErrTemporary / unknown: exponential backoff retry -func (m *Manager) sendWithRetry(ctx context.Context, name string, w *channelWorker, msg bus.OutboundMessage) { +func (m *Manager) sendWithRetry( + ctx context.Context, + name string, + w *channelWorker, + msg bus.OutboundMessage, +) { // Rate limit: wait for token if err := w.limiter.Wait(ctx); err != nil { // ctx canceled, shutting down @@ -663,7 +679,10 @@ func (m *Manager) sendWithRetry(ctx context.Context, name string, w *channelWork } // ErrTemporary or unknown error ā€” exponential backoff - backoff := min(time.Duration(float64(baseBackoff)*math.Pow(2, float64(attempt))), maxBackoff) + backoff := min( + time.Duration(float64(baseBackoff)*math.Pow(2, float64(attempt))), + maxBackoff, + ) select { case <-time.After(backoff): case <-ctx.Done(): @@ -788,12 +807,21 @@ func (m *Manager) runMediaWorker(ctx context.Context, name string, w *channelWor // sendMediaWithRetry sends a media message through the channel with rate limiting and // retry logic. If the channel does not implement MediaSender, it silently skips. -func (m *Manager) sendMediaWithRetry(ctx context.Context, name string, w *channelWorker, msg bus.OutboundMediaMessage) { +func (m *Manager) sendMediaWithRetry( + ctx context.Context, + name string, + w *channelWorker, + msg bus.OutboundMediaMessage, +) { ms, ok := w.ch.(MediaSender) if !ok { - logger.DebugCF("channels", "Channel does not support MediaSender, skipping media", map[string]any{ - "channel": name, - }) + logger.DebugCF( + "channels", + "Channel does not support MediaSender, skipping media", + map[string]any{ + "channel": name, + }, + ) return } @@ -830,7 +858,10 @@ func (m *Manager) sendMediaWithRetry(ctx context.Context, name string, w *channe } // ErrTemporary or unknown error ā€” exponential backoff - backoff := min(time.Duration(float64(baseBackoff)*math.Pow(2, float64(attempt))), maxBackoff) + backoff := min( + time.Duration(float64(baseBackoff)*math.Pow(2, float64(attempt))), + maxBackoff, + ) select { case <-time.After(backoff): case <-ctx.Done(): diff --git a/pkg/channels/mqtt/init.go b/pkg/channels/mqtt/init.go index c5f77713f..ef0acf181 100644 --- a/pkg/channels/mqtt/init.go +++ b/pkg/channels/mqtt/init.go @@ -7,10 +7,17 @@ import ( ) func init() { + channels.RegisterFactory("mqtt", func(cfg *config.Config, b *bus.MessageBus) (channels.Channel, error) { + if !cfg.Channels.MQTT.Enabled { + return nil, nil + } + return NewMQTTChannel(cfg.Channels.MQTT, b) + }) -} \ No newline at end of file + +} diff --git a/pkg/channels/mqtt/mqtt.go b/pkg/channels/mqtt/mqtt.go index 08809d16e..fdaaabd90 100644 --- a/pkg/channels/mqtt/mqtt.go +++ b/pkg/channels/mqtt/mqtt.go @@ -11,7 +11,6 @@ import ( "time" mqtt "github.com/eclipse/paho.mqtt.golang" - "github.com/sipeed/picoclaw/pkg/bus" "github.com/sipeed/picoclaw/pkg/channels" "github.com/sipeed/picoclaw/pkg/config" @@ -87,7 +86,7 @@ func (c *MQTTChannel) Start(ctx context.Context) error { tlsConfig := &tls.Config{ InsecureSkipVerify: false, } - + // Load CA certificate if provided if c.config.TLSCA != "" { caCert, err := os.ReadFile(c.config.TLSCA) @@ -100,7 +99,7 @@ func (c *MQTTChannel) Start(ctx context.Context) error { } tlsConfig.RootCAs = caCertPool } - + // Load client certificate and key if provided if c.config.TLSCert != "" && c.config.TLSKey != "" { cert, err := tls.LoadX509KeyPair(c.config.TLSCert, c.config.TLSKey) @@ -109,7 +108,7 @@ func (c *MQTTChannel) Start(ctx context.Context) error { } tlsConfig.Certificates = []tls.Certificate{cert} } - + opts.SetTLSConfig(tlsConfig) } @@ -127,7 +126,7 @@ func (c *MQTTChannel) Start(ctx context.Context) error { // Set connect handler opts.SetOnConnectHandler(func(client mqtt.Client) { logger.InfoC("mqtt", "Connected to MQTT broker") - + // Subscribe to topics after successful connection var subscriptionErrors []string for _, topic := range c.config.SubscribeTopics { @@ -144,7 +143,7 @@ func (c *MQTTChannel) Start(ctx context.Context) error { }) } } - + // Log subscription errors summary if len(subscriptionErrors) > 0 { logger.ErrorCF("mqtt", "Subscription errors occurred", map[string]any{ @@ -164,13 +163,13 @@ func (c *MQTTChannel) Start(ctx context.Context) error { } logger.InfoCF("mqtt", "Connected to MQTT broker", map[string]any{ - "broker": c.config.Broker, + "broker": c.config.Broker, "client_id": c.config.ClientID, }) c.SetRunning(true) logger.InfoC("mqtt", "MQTT channel started") - + return nil } @@ -191,7 +190,6 @@ func (c *MQTTChannel) Stop(ctx context.Context) error { return nil } - // onMessage handles incoming MQTT messages. func (c *MQTTChannel) onMessage(client mqtt.Client, msg mqtt.Message) { logger.DebugCF("mqtt", "Received message", map[string]any{ @@ -200,7 +198,7 @@ func (c *MQTTChannel) onMessage(client mqtt.Client, msg mqtt.Message) { }) var content string - + // Check if subscribe_json_key is configured if c.config.SubscribeJSONKey != nil && *c.config.SubscribeJSONKey != "" { // Parse as JSON and extract the specified key @@ -277,8 +275,8 @@ func (c *MQTTChannel) onMessage(client mqtt.Client, msg mqtt.Message) { messageID := fmt.Sprintf("mqtt-%d", time.Now().UnixNano()) metadata := map[string]string{ - "platform": "mqtt", - "topic": msg.Topic(), + "platform": "mqtt", + "topic": msg.Topic(), "reply_topic": replyTopic, } @@ -313,7 +311,7 @@ func (c *MQTTChannel) Send(ctx context.Context, msg bus.OutboundMessage) error { if replyTopic == "" { return fmt.Errorf("reply topic is empty and no default configured") } - + var payload []byte var err error diff --git a/pkg/config/config.go b/pkg/config/config.go index b4138eb45..dcc0715d3 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -594,26 +594,26 @@ type IRCConfig struct { } type MQTTConfig struct { - Enabled bool `json:"enabled" env:"PICOCLAW_CHANNELS_MQTT_ENABLED"` - Broker string `json:"broker" env:"PICOCLAW_CHANNELS_MQTT_BROKER"` - ClientID string `json:"client_id" env:"PICOCLAW_CHANNELS_MQTT_CLIENT_ID"` - Username string `json:"username" env:"PICOCLAW_CHANNELS_MQTT_USERNAME"` - Password string `json:"password" env:"PICOCLAW_CHANNELS_MQTT_PASSWORD"` - SubscribeTopics []string `json:"subscribe_topics" env:"PICOCLAW_CHANNELS_MQTT_SUBSCRIBE_TOPICS"` + Enabled bool `json:"enabled" env:"PICOCLAW_CHANNELS_MQTT_ENABLED"` + Broker string `json:"broker" env:"PICOCLAW_CHANNELS_MQTT_BROKER"` + ClientID string `json:"client_id" env:"PICOCLAW_CHANNELS_MQTT_CLIENT_ID"` + Username string `json:"username" env:"PICOCLAW_CHANNELS_MQTT_USERNAME"` + Password string `json:"password" env:"PICOCLAW_CHANNELS_MQTT_PASSWORD"` + SubscribeTopics []string `json:"subscribe_topics" env:"PICOCLAW_CHANNELS_MQTT_SUBSCRIBE_TOPICS"` SubscribeJSONKey *string `json:"subscribe_json_key,omitempty"` - ReplyTopic string `json:"reply_topic" env:"PICOCLAW_CHANNELS_MQTT_REPLY_TOPIC"` + ReplyTopic string `json:"reply_topic" env:"PICOCLAW_CHANNELS_MQTT_REPLY_TOPIC"` ReplyJSONKey *string `json:"reply_json_key,omitempty"` - TLS bool `json:"tls" env:"PICOCLAW_CHANNELS_MQTT_TLS"` - TLSCA string `json:"tls_ca" env:"PICOCLAW_CHANNELS_MQTT_TLS_CA"` - TLSCert string `json:"tls_cert" env:"PICOCLAW_CHANNELS_MQTT_TLS_CERT"` - TLSKey string `json:"tls_key" env:"PICOCLAW_CHANNELS_MQTT_TLS_KEY"` - QoS int `json:"qos" env:"PICOCLAW_CHANNELS_MQTT_QOS"` - Retain bool `json:"retain" env:"PICOCLAW_CHANNELS_MQTT_RETAIN"` - Prefix string `json:"prefix" env:"PICOCLAW_CHANNELS_MQTT_PREFIX"` - Instruction string `json:"instruction" env:"PICOCLAW_CHANNELS_MQTT_INSTRUCTION"` - AllowFrom FlexibleStringSlice `json:"allow_from" env:"PICOCLAW_CHANNELS_MQTT_ALLOW_FROM"` + TLS bool `json:"tls" env:"PICOCLAW_CHANNELS_MQTT_TLS"` + TLSCA string `json:"tls_ca" env:"PICOCLAW_CHANNELS_MQTT_TLS_CA"` + TLSCert string `json:"tls_cert" env:"PICOCLAW_CHANNELS_MQTT_TLS_CERT"` + TLSKey string `json:"tls_key" env:"PICOCLAW_CHANNELS_MQTT_TLS_KEY"` + QoS int `json:"qos" env:"PICOCLAW_CHANNELS_MQTT_QOS"` + Retain bool `json:"retain" env:"PICOCLAW_CHANNELS_MQTT_RETAIN"` + Prefix string `json:"prefix" env:"PICOCLAW_CHANNELS_MQTT_PREFIX"` + Instruction string `json:"instruction" env:"PICOCLAW_CHANNELS_MQTT_INSTRUCTION"` + AllowFrom FlexibleStringSlice `json:"allow_from" env:"PICOCLAW_CHANNELS_MQTT_ALLOW_FROM"` GroupTrigger GroupTriggerConfig `json:"group_trigger,omitempty"` - ReasoningChannelID string `json:"reasoning_channel_id" env:"PICOCLAW_CHANNELS_MQTT_REASONING_CHANNEL_ID"` + ReasoningChannelID string `json:"reasoning_channel_id" env:"PICOCLAW_CHANNELS_MQTT_REASONING_CHANNEL_ID"` } type HeartbeatConfig struct { @@ -808,9 +808,9 @@ type SearXNGConfig struct { } type GLMSearchConfig struct { - Enabled bool `json:"enabled" env:"PICOCLAW_TOOLS_WEB_GLM_ENABLED"` - APIKey string `json:"api_key" env:"PICOCLAW_TOOLS_WEB_GLM_API_KEY"` - BaseURL string `json:"base_url" env:"PICOCLAW_TOOLS_WEB_GLM_BASE_URL"` + Enabled bool `json:"enabled" env:"PICOCLAW_TOOLS_WEB_GLM_ENABLED"` + APIKey string `json:"api_key" env:"PICOCLAW_TOOLS_WEB_GLM_API_KEY"` + BaseURL string `json:"base_url" env:"PICOCLAW_TOOLS_WEB_GLM_BASE_URL"` // SearchEngine specifies the search backend: "search_std" (default), // "search_pro", "search_pro_sogou", or "search_pro_quark". SearchEngine string `json:"search_engine" env:"PICOCLAW_TOOLS_WEB_GLM_SEARCH_ENGINE"` @@ -825,6 +825,7 @@ type BaiduSearchConfig struct { } type WebToolsConfig struct { + ToolConfig ` envPrefix:"PICOCLAW_TOOLS_WEB_"` Brave BraveConfig ` json:"brave"` Tavily TavilyConfig ` json:"tavily"` @@ -833,18 +834,19 @@ type WebToolsConfig struct { SearXNG SearXNGConfig ` json:"searxng"` GLMSearch GLMSearchConfig ` json:"glm_search"` BaiduSearch BaiduSearchConfig ` json:"baidu_search"` + // PreferNative controls whether to use provider-native web search when // the active LLM supports it (e.g. OpenAI web_search_preview). When true, // the client-side web_search tool is hidden to avoid duplicate search surfaces, // and the provider's built-in search is used instead. Falls back to client-side // search when the provider does not support native search. - PreferNative bool `json:"prefer_native" env:"PICOCLAW_TOOLS_WEB_PREFER_NATIVE"` + PreferNative bool ` json:"prefer_native" env:"PICOCLAW_TOOLS_WEB_PREFER_NATIVE"` // Proxy is an optional proxy URL for web tools (http/https/socks5/socks5h). // For authenticated proxies, prefer HTTP_PROXY/HTTPS_PROXY env vars instead of embedding credentials in config. - Proxy string `json:"proxy,omitempty" env:"PICOCLAW_TOOLS_WEB_PROXY"` - FetchLimitBytes int64 `json:"fetch_limit_bytes,omitempty" env:"PICOCLAW_TOOLS_WEB_FETCH_LIMIT_BYTES"` - Format string `json:"format,omitempty" env:"PICOCLAW_TOOLS_WEB_FORMAT"` - PrivateHostWhitelist FlexibleStringSlice `json:"private_host_whitelist,omitempty" env:"PICOCLAW_TOOLS_WEB_PRIVATE_HOST_WHITELIST"` + Proxy string ` json:"proxy,omitempty" env:"PICOCLAW_TOOLS_WEB_PROXY"` + FetchLimitBytes int64 ` json:"fetch_limit_bytes,omitempty" env:"PICOCLAW_TOOLS_WEB_FETCH_LIMIT_BYTES"` + Format string ` json:"format,omitempty" env:"PICOCLAW_TOOLS_WEB_FORMAT"` + PrivateHostWhitelist FlexibleStringSlice ` json:"private_host_whitelist,omitempty" env:"PICOCLAW_TOOLS_WEB_PRIVATE_HOST_WHITELIST"` } type CronToolsConfig struct { @@ -959,10 +961,10 @@ type MCPServerConfig struct { // MCPConfig defines configuration for all MCP servers type MCPConfig struct { - ToolConfig ` envPrefix:"PICOCLAW_TOOLS_MCP_"` + ToolConfig ` envPrefix:"PICOCLAW_TOOLS_MCP_"` Discovery ToolDiscoveryConfig ` json:"discovery"` // Servers is a map of server name to server configuration - Servers map[string]MCPServerConfig `json:"servers,omitempty"` + Servers map[string]MCPServerConfig ` json:"servers,omitempty"` } func LoadConfig(path string) (*Config, error) { diff --git a/pkg/gateway/gateway.go b/pkg/gateway/gateway.go index 8a3db3fee..6643390a8 100644 --- a/pkg/gateway/gateway.go +++ b/pkg/gateway/gateway.go @@ -45,20 +45,29 @@ import ( ) const ( - serviceShutdownTimeout = 30 * time.Second - providerReloadTimeout = 30 * time.Second + serviceShutdownTimeout = 30 * time.Second + + providerReloadTimeout = 30 * time.Second + gracefulShutdownTimeout = 15 * time.Second ) type services struct { - CronService *cron.CronService + CronService *cron.CronService + HeartbeatService *heartbeat.HeartbeatService - MediaStore media.MediaStore - ChannelManager *channels.Manager - DeviceService *devices.Service - HealthServer *health.Server + + MediaStore media.MediaStore + + ChannelManager *channels.Manager + + DeviceService *devices.Service + + HealthServer *health.Server + manualReloadChan chan struct{} - reloading atomic.Bool + + reloading atomic.Bool } type startupBlockedProvider struct { @@ -66,24 +75,47 @@ type startupBlockedProvider struct { } func (p *startupBlockedProvider) Chat( + _ context.Context, + _ []providers.Message, + _ []providers.ToolDefinition, + _ string, + _ map[string]any, + ) (*providers.LLMResponse, error) { + return nil, fmt.Errorf("%s", p.reason) + } func (p *startupBlockedProvider) GetDefaultModel() string { + return "" + } // Run starts the gateway runtime using the configuration loaded from configPath. + func Run(debug bool, configPath string, allowEmptyStartup bool) error { + + if debug { + + logger.SetLevel(logger.DEBUG) + + fmt.Println("šŸ” Debug mode enabled") + + } + cfg, err := config.LoadConfig(configPath) + if err != nil { + return fmt.Errorf("error loading config: %w", err) + } logger.SetLevelFromString(cfg.Gateway.LogLevel) @@ -94,296 +126,505 @@ func Run(debug bool, configPath string, allowEmptyStartup bool) error { } provider, modelID, err := createStartupProvider(cfg, allowEmptyStartup) + if err != nil { + return fmt.Errorf("error creating provider: %w", err) + } if modelID != "" { + cfg.Agents.Defaults.ModelName = modelID + } msgBus := bus.NewMessageBus() + agentLoop := agent.NewAgentLoop(cfg, msgBus, provider) fmt.Println("\nšŸ“¦ Agent Status:") + startupInfo := agentLoop.GetStartupInfo() + toolsInfo := startupInfo["tools"].(map[string]any) + skillsInfo := startupInfo["skills"].(map[string]any) + fmt.Printf(" ā€¢ Tools: %d loaded\n", toolsInfo["count"]) + fmt.Printf(" ā€¢ Skills: %d/%d available\n", skillsInfo["available"], skillsInfo["total"]) logger.InfoCF("agent", "Agent initialized", + map[string]any{ - "tools_count": toolsInfo["count"], - "skills_total": skillsInfo["total"], + + "tools_count": toolsInfo["count"], + + "skills_total": skillsInfo["total"], + "skills_available": skillsInfo["available"], }) runningServices, err := setupAndStartServices(cfg, agentLoop, msgBus) + if err != nil { + return err + } // Setup manual reload channel for /reload endpoint + manualReloadChan := make(chan struct{}, 1) + runningServices.manualReloadChan = manualReloadChan + reloadTrigger := func() error { + if !runningServices.reloading.CompareAndSwap(false, true) { + return fmt.Errorf("reload already in progress") + } + select { + case manualReloadChan <- struct{}{}: + return nil + default: + // Should not happen, but reset flag if channel is full + runningServices.reloading.Store(false) + return fmt.Errorf("reload already queued") + } + } + runningServices.HealthServer.SetReloadFunc(reloadTrigger) + agentLoop.SetReloadFunc(reloadTrigger) fmt.Printf("āœ“ Gateway started on %s:%d\n", cfg.Gateway.Host, cfg.Gateway.Port) + fmt.Println("Press Ctrl+C to stop") ctx, cancel := context.WithCancel(context.Background()) + defer cancel() go agentLoop.Run(ctx) var configReloadChan <-chan *config.Config + stopWatch := func() {} + if cfg.Gateway.HotReload { + configReloadChan, stopWatch = setupConfigWatcherPolling(configPath, debug) + logger.Info("Config hot reload enabled") + } + defer stopWatch() sigChan := make(chan os.Signal, 1) + signal.Notify(sigChan, os.Interrupt, syscall.SIGTERM) for { + select { + case <-sigChan: + logger.Info("Shutting down...") + shutdownGateway(runningServices, agentLoop, provider, true) + return nil + case newCfg := <-configReloadChan: + if !runningServices.reloading.CompareAndSwap(false, true) { + logger.Warn("Config reload skipped: another reload is in progress") + continue + } + err := executeReload(ctx, agentLoop, newCfg, &provider, runningServices, msgBus, allowEmptyStartup) + if err != nil { + logger.Errorf("Config reload failed: %v", err) + } + case <-manualReloadChan: + logger.Info("Manual reload triggered via /reload endpoint") + newCfg, err := config.LoadConfig(configPath) + if err != nil { + logger.Errorf("Error loading config for manual reload: %v", err) + runningServices.reloading.Store(false) + continue + } + if err = newCfg.ValidateModelList(); err != nil { + logger.Errorf("Config validation failed: %v", err) + runningServices.reloading.Store(false) + continue + } + err = executeReload(ctx, agentLoop, newCfg, &provider, runningServices, msgBus, allowEmptyStartup) + if err != nil { + logger.Errorf("Manual reload failed: %v", err) + } else { + logger.Info("Manual reload completed successfully") + } + } + } + } func executeReload( + ctx context.Context, + agentLoop *agent.AgentLoop, + newCfg *config.Config, + provider *providers.LLMProvider, + runningServices *services, + msgBus *bus.MessageBus, + allowEmptyStartup bool, + ) error { + defer runningServices.reloading.Store(false) + return handleConfigReload(ctx, agentLoop, newCfg, provider, runningServices, msgBus, allowEmptyStartup) + } func createStartupProvider( + cfg *config.Config, + allowEmptyStartup bool, + ) (providers.LLMProvider, string, error) { + modelName := cfg.Agents.Defaults.GetModelName() + if modelName == "" && allowEmptyStartup { + reason := "no default model configured; gateway started in limited mode" + fmt.Printf("āš  Warning: %s\n", reason) + logger.WarnCF("gateway", "Gateway started without default model", map[string]any{ + "limited_mode": true, }) + return &startupBlockedProvider{reason: reason}, "", nil + } return providers.CreateProvider(cfg) + } func setupAndStartServices( + cfg *config.Config, + agentLoop *agent.AgentLoop, + msgBus *bus.MessageBus, + ) (*services, error) { + runningServices := &services{} execTimeout := time.Duration(cfg.Tools.Cron.ExecTimeoutMinutes) * time.Minute + var err error + runningServices.CronService, err = setupCronTool( + agentLoop, + msgBus, + cfg.WorkspacePath(), + cfg.Agents.Defaults.RestrictToWorkspace, + execTimeout, + cfg, ) + if err != nil { + return nil, fmt.Errorf("error setting up cron service: %w", err) + } + if err = runningServices.CronService.Start(); err != nil { + return nil, fmt.Errorf("error starting cron service: %w", err) + } + fmt.Println("āœ“ Cron service started") runningServices.HeartbeatService = heartbeat.NewHeartbeatService( + cfg.WorkspacePath(), + cfg.Heartbeat.Interval, + cfg.Heartbeat.Enabled, ) + runningServices.HeartbeatService.SetBus(msgBus) + runningServices.HeartbeatService.SetHandler(createHeartbeatHandler(agentLoop)) + if err = runningServices.HeartbeatService.Start(); err != nil { + return nil, fmt.Errorf("error starting heartbeat service: %w", err) + } + fmt.Println("āœ“ Heartbeat service started") runningServices.MediaStore = media.NewFileMediaStoreWithCleanup(media.MediaCleanerConfig{ - Enabled: cfg.Tools.MediaCleanup.Enabled, - MaxAge: time.Duration(cfg.Tools.MediaCleanup.MaxAge) * time.Minute, + + Enabled: cfg.Tools.MediaCleanup.Enabled, + + MaxAge: time.Duration(cfg.Tools.MediaCleanup.MaxAge) * time.Minute, + Interval: time.Duration(cfg.Tools.MediaCleanup.Interval) * time.Minute, }) + if fms, ok := runningServices.MediaStore.(*media.FileMediaStore); ok { + fms.Start() + } runningServices.ChannelManager, err = channels.NewManager(cfg, msgBus, runningServices.MediaStore) + if err != nil { + if fms, ok := runningServices.MediaStore.(*media.FileMediaStore); ok { + fms.Stop() + } + return nil, fmt.Errorf("error creating channel manager: %w", err) + } agentLoop.SetChannelManager(runningServices.ChannelManager) + agentLoop.SetMediaStore(runningServices.MediaStore) if transcriber := voice.DetectTranscriber(cfg); transcriber != nil { + agentLoop.SetTranscriber(transcriber) + logger.InfoCF("voice", "Transcription enabled (agent-level)", map[string]any{"provider": transcriber.Name()}) + } enabledChannels := runningServices.ChannelManager.GetEnabledChannels() + if len(enabledChannels) > 0 { + fmt.Printf("āœ“ Channels enabled: %s\n", enabledChannels) + } else { + fmt.Println("āš  Warning: No channels enabled") + } addr := fmt.Sprintf("%s:%d", cfg.Gateway.Host, cfg.Gateway.Port) + runningServices.HealthServer = health.NewServer(cfg.Gateway.Host, cfg.Gateway.Port) + runningServices.ChannelManager.SetupHTTPServer(addr, runningServices.HealthServer) if err = runningServices.ChannelManager.StartAll(context.Background()); err != nil { + return nil, fmt.Errorf("error starting channels: %w", err) + } fmt.Printf( + "āœ“ Health endpoints available at http://%s:%d/health, /ready and /reload (POST)\n", + cfg.Gateway.Host, + cfg.Gateway.Port, ) stateManager := state.NewManager(cfg.WorkspacePath()) + runningServices.DeviceService = devices.NewService(devices.Config{ - Enabled: cfg.Devices.Enabled, + + Enabled: cfg.Devices.Enabled, + MonitorUSB: cfg.Devices.MonitorUSB, }, stateManager) + runningServices.DeviceService.SetBus(msgBus) + if err = runningServices.DeviceService.Start(context.Background()); err != nil { + logger.ErrorCF("device", "Error starting device service", map[string]any{"error": err.Error()}) + } else if cfg.Devices.Enabled { + fmt.Println("āœ“ Device event service started") + } return runningServices, nil + } func stopAndCleanupServices(runningServices *services, shutdownTimeout time.Duration, isReload bool) { shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), shutdownTimeout) + defer shutdownCancel() // reload should not stop channel manager if !isReload && runningServices.ChannelManager != nil { runningServices.ChannelManager.StopAll(shutdownCtx) + } + if runningServices.DeviceService != nil { + runningServices.DeviceService.Stop() + } + if runningServices.HeartbeatService != nil { + runningServices.HeartbeatService.Stop() + } + if runningServices.CronService != nil { + runningServices.CronService.Stop() + } + if runningServices.MediaStore != nil { + if fms, ok := runningServices.MediaStore.(*media.FileMediaStore); ok { + fms.Stop() + } + } + } func shutdownGateway( + runningServices *services, + agentLoop *agent.AgentLoop, + provider providers.LLMProvider, + fullShutdown bool, + ) { + if cp, ok := provider.(providers.StatefulProvider); ok && fullShutdown { + cp.Close() + } stopAndCleanupServices(runningServices, gracefulShutdownTimeout, false) agentLoop.Stop() + agentLoop.Close() logger.Info("āœ“ Gateway stopped") + } func handleConfigReload( + ctx context.Context, + al *agent.AgentLoop, + newCfg *config.Config, + providerRef *providers.LLMProvider, + runningServices *services, + msgBus *bus.MessageBus, + allowEmptyStartup bool, + ) error { + logger.Info("šŸ”„ Config file changed, reloading...") newModel := newCfg.Agents.Defaults.ModelName + if newModel == "" { + newModel = newCfg.Agents.Defaults.Model + } logger.Infof(" New model is '%s', recreating provider...", newModel) @@ -392,269 +633,447 @@ func handleConfigReload( stopAndCleanupServices(runningServices, serviceShutdownTimeout, true) newProvider, newModelID, err := createStartupProvider(newCfg, allowEmptyStartup) + if err != nil { + logger.Errorf(" āš  Error creating new provider: %v", err) + logger.Warn(" Attempting to restart services with old provider and config...") + if restartErr := restartServices(al, runningServices, msgBus); restartErr != nil { + logger.Errorf(" āš  Failed to restart services: %v", restartErr) + } + return fmt.Errorf("error creating new provider: %w", err) + } if newModelID != "" { + newCfg.Agents.Defaults.ModelName = newModelID + } reloadCtx, reloadCancel := context.WithTimeout(context.Background(), providerReloadTimeout) + defer reloadCancel() if err := al.ReloadProviderAndConfig(reloadCtx, newProvider, newCfg); err != nil { + logger.Errorf(" āš  Error reloading agent loop: %v", err) + if cp, ok := newProvider.(providers.StatefulProvider); ok { + cp.Close() + } + logger.Warn(" Attempting to restart services with old provider and config...") + if restartErr := restartServices(al, runningServices, msgBus); restartErr != nil { + logger.Errorf(" āš  Failed to restart services: %v", restartErr) + } + return fmt.Errorf("error reloading agent loop: %w", err) + } *providerRef = newProvider logger.Info(" Restarting all services with new configuration...") + if err := restartServices(al, runningServices, msgBus); err != nil { + logger.Errorf(" āš  Error restarting services: %v", err) + return fmt.Errorf("error restarting services: %w", err) + } logger.Info(" āœ“ Provider, configuration, and services reloaded successfully (thread-safe)") + return nil + } func restartServices( + al *agent.AgentLoop, + runningServices *services, + msgBus *bus.MessageBus, + ) error { + cfg := al.GetConfig() execTimeout := time.Duration(cfg.Tools.Cron.ExecTimeoutMinutes) * time.Minute + var err error + runningServices.CronService, err = setupCronTool( + al, + msgBus, + cfg.WorkspacePath(), + cfg.Agents.Defaults.RestrictToWorkspace, + execTimeout, + cfg, ) + if err != nil { + return fmt.Errorf("error restarting cron service: %w", err) + } + if err = runningServices.CronService.Start(); err != nil { + return fmt.Errorf("error restarting cron service: %w", err) + } + fmt.Println(" āœ“ Cron service restarted") runningServices.HeartbeatService = heartbeat.NewHeartbeatService( + cfg.WorkspacePath(), + cfg.Heartbeat.Interval, + cfg.Heartbeat.Enabled, ) + runningServices.HeartbeatService.SetBus(msgBus) + runningServices.HeartbeatService.SetHandler(createHeartbeatHandler(al)) + if err = runningServices.HeartbeatService.Start(); err != nil { + return fmt.Errorf("error restarting heartbeat service: %w", err) + } + fmt.Println(" āœ“ Heartbeat service restarted") runningServices.MediaStore = media.NewFileMediaStoreWithCleanup(media.MediaCleanerConfig{ - Enabled: cfg.Tools.MediaCleanup.Enabled, - MaxAge: time.Duration(cfg.Tools.MediaCleanup.MaxAge) * time.Minute, + + Enabled: cfg.Tools.MediaCleanup.Enabled, + + MaxAge: time.Duration(cfg.Tools.MediaCleanup.MaxAge) * time.Minute, + Interval: time.Duration(cfg.Tools.MediaCleanup.Interval) * time.Minute, }) + if fms, ok := runningServices.MediaStore.(*media.FileMediaStore); ok { + fms.Start() + } + al.SetMediaStore(runningServices.MediaStore) runningServices.ChannelManager, err = channels.NewManager(cfg, msgBus, runningServices.MediaStore) + if err != nil { + return fmt.Errorf("error recreating channel manager: %w", err) + } + al.SetChannelManager(runningServices.ChannelManager) enabledChannels := runningServices.ChannelManager.GetEnabledChannels() + if len(enabledChannels) > 0 { + fmt.Printf(" āœ“ Channels enabled: %s\n", enabledChannels) + } else { + fmt.Println(" āš  Warning: No channels enabled") + } addr := fmt.Sprintf("%s:%d", cfg.Gateway.Host, cfg.Gateway.Port) + // Reuse existing HealthServer to preserve reloadFunc + if runningServices.HealthServer == nil { + runningServices.HealthServer = health.NewServer(cfg.Gateway.Host, cfg.Gateway.Port) + } + runningServices.ChannelManager.SetupHTTPServer(addr, runningServices.HealthServer) if err = runningServices.ChannelManager.Reload(context.Background(), cfg); err != nil { return fmt.Errorf("error reload channels: %w", err) } + fmt.Println(" āœ“ Channels restarted.") stateManager := state.NewManager(cfg.WorkspacePath()) + runningServices.DeviceService = devices.NewService(devices.Config{ - Enabled: cfg.Devices.Enabled, + + Enabled: cfg.Devices.Enabled, + MonitorUSB: cfg.Devices.MonitorUSB, }, stateManager) + runningServices.DeviceService.SetBus(msgBus) + if err := runningServices.DeviceService.Start(context.Background()); err != nil { + logger.WarnCF("device", "Failed to restart device service", map[string]any{"error": err.Error()}) + } else if cfg.Devices.Enabled { + fmt.Println(" āœ“ Device event service restarted") + } transcriber := voice.DetectTranscriber(cfg) + al.SetTranscriber(transcriber) + if transcriber != nil { + logger.InfoCF("voice", "Transcription re-enabled (agent-level)", map[string]any{"provider": transcriber.Name()}) + } else { + logger.InfoCF("voice", "Transcription disabled", nil) + } return nil + } func setupConfigWatcherPolling(configPath string, debug bool) (chan *config.Config, func()) { + configChan := make(chan *config.Config, 1) + stop := make(chan struct{}) + var wg sync.WaitGroup wg.Add(1) + go func() { + defer wg.Done() lastModTime := getFileModTime(configPath) + lastSize := getFileSize(configPath) ticker := time.NewTicker(2 * time.Second) + defer ticker.Stop() for { + select { + case <-ticker.C: + currentModTime := getFileModTime(configPath) + currentSize := getFileSize(configPath) if currentModTime.After(lastModTime) || currentSize != lastSize { + if debug { + logger.Debugf("šŸ” Config file change detected") + } time.Sleep(500 * time.Millisecond) lastModTime = currentModTime + lastSize = currentSize newCfg, err := config.LoadConfig(configPath) + if err != nil { + logger.Errorf("āš  Error loading new config: %v", err) + logger.Warn(" Using previous valid config") + continue + } if err := newCfg.ValidateModelList(); err != nil { + logger.Errorf(" āš  New config validation failed: %v", err) + logger.Warn(" Using previous valid config") + continue + } logger.Info("āœ“ Config file validated and loaded") select { + case configChan <- newCfg: + default: + logger.Warn("āš  Previous config reload still in progress, skipping") + } + } + case <-stop: + return + } + } + }() stopFunc := func() { + close(stop) + wg.Wait() + } return configChan, stopFunc + } func getFileModTime(path string) time.Time { + info, err := os.Stat(path) + if err != nil { + return time.Time{} + } + return info.ModTime() + } func getFileSize(path string) int64 { + info, err := os.Stat(path) + if err != nil { + return 0 + } + return info.Size() + } func setupCronTool( + agentLoop *agent.AgentLoop, + msgBus *bus.MessageBus, + workspace string, + restrict bool, + execTimeout time.Duration, + cfg *config.Config, + ) (*cron.CronService, error) { + cronStorePath := filepath.Join(workspace, "cron", "jobs.json") cronService := cron.NewCronService(cronStorePath, nil) var cronTool *tools.CronTool + if cfg.Tools.IsToolEnabled("cron") { + var err error + cronTool, err = tools.NewCronTool(cronService, agentLoop, msgBus, workspace, restrict, execTimeout, cfg) + if err != nil { + return nil, fmt.Errorf("critical error during CronTool initialization: %w", err) + } agentLoop.RegisterTool(cronTool) + } if cronTool != nil { + cronService.SetOnJob(func(job *cron.CronJob) (string, error) { + result := cronTool.ExecuteJob(context.Background(), job) + return result, nil + }) + } return cronService, nil + } func createHeartbeatHandler(agentLoop *agent.AgentLoop) func(prompt, channel, chatID string) *tools.ToolResult { + return func(prompt, channel, chatID string) *tools.ToolResult { + if channel == "" || chatID == "" { + channel, chatID = "cli", "direct" + } response, err := agentLoop.ProcessHeartbeat(context.Background(), prompt, channel, chatID) + if err != nil { + return tools.ErrorResult(fmt.Sprintf("Heartbeat error: %v", err)) + } + if response == "HEARTBEAT_OK" { + return tools.SilentResult("Heartbeat OK") + } + return tools.SilentResult(response) + } + }