From 3335ceb9b7789c63759cff02325d1c13551f2973 Mon Sep 17 00:00:00 2001 From: Stefan Rinke Date: Sun, 15 Feb 2026 18:20:50 +0100 Subject: [PATCH] Added a remote admin interface that basically can reconfigure the agent --- README.md | 44 +++++++++++ cmd/picoclaw/main.go | 95 ++++++++++++++++++++--- config/config.example.json | 3 +- pkg/agent/loop.go | 19 +++++ pkg/channels/webui.go | 150 ++++++++++++++++++++++++++++++++++--- pkg/config/config.go | 2 + pkg/migrate/config.go | 3 + 7 files changed, 297 insertions(+), 19 deletions(-) diff --git a/README.md b/README.md index f1f144009..3ea63d024 100644 --- a/README.md +++ b/README.md @@ -796,6 +796,50 @@ If you set `gateway.token`, open: The listen address is controlled by `gateway.bind` (`local`, `tailnet`, `all`). +### Admin API (Config Update + Graceful Restart) + +The gateway also exposes an **API-only** management interface for remote administration. + +It supports: + +- **Replace config**: `PUT /admin/config` +- **Graceful drain + exit(0)** (so Docker/systemd can restart it): `POST /admin/drain-exit` + +**Authentication** + +Set `gateway.admin_token` (or `PICOCLAW_GATEWAY_ADMIN_TOKEN`) and pass it via: + +`Authorization: Bearer ` + +If `gateway.admin_token` is empty, the admin API will always return `401 Unauthorized`. + +**Writable config required** + +The gateway writes to its normal config path (e.g. `~/.picoclaw/config.json`, or `/root/.picoclaw/config.json` in Docker). +If that file is mounted read-only, the config update endpoint will fail with **"config is not writable"**. + +**Examples** + +Replace the full config: + +```bash +curl -X PUT \ + -H "Authorization: Bearer " \ + -H "Content-Type: application/json" \ + --data-binary @config.json \ + http://:18790/admin/config +``` + +Trigger graceful drain and restart (default timeout 30s): + +```bash +curl -X POST \ + -H "Authorization: Bearer " \ + -H "Content-Type: application/json" \ + -d '{"timeout_seconds":30}' \ + http://:18790/admin/drain-exit +``` + ## CLI Reference | Command | Description | diff --git a/cmd/picoclaw/main.go b/cmd/picoclaw/main.go index 8fb53faf3..8ae91ede0 100644 --- a/cmd/picoclaw/main.go +++ b/cmd/picoclaw/main.go @@ -10,6 +10,7 @@ import ( "bufio" "context" "embed" + "encoding/json" "fmt" "io" "io/fs" @@ -592,6 +593,59 @@ func gatewayCmd() { os.Exit(1) } + configPath := getConfigPath() + + saveConfigRawAtomic := func(path string, raw []byte) error { + // Validate incoming JSON against Config schema first. + validated := config.DefaultConfig() + if err := json.Unmarshal(raw, validated); err != nil { + return fmt.Errorf("invalid config json: %w", err) + } + + // Ensure config dir exists. + dir := filepath.Dir(path) + if err := os.MkdirAll(dir, 0755); err != nil { + if os.IsPermission(err) { + return fmt.Errorf("config is not writable: %w", err) + } + return err + } + + // Atomic write: write to temp file then rename. + tmp := path + ".tmp" + f, err := os.OpenFile(tmp, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0644) + if err != nil { + if os.IsPermission(err) { + return fmt.Errorf("config is not writable") + } + return err + } + _, werr := f.Write(raw) + err = f.Close() + if werr != nil { + _ = os.Remove(tmp) + if os.IsPermission(werr) { + return fmt.Errorf("config is not writable") + } + return werr + } + if err != nil { + _ = os.Remove(tmp) + if os.IsPermission(err) { + return fmt.Errorf("config is not writable") + } + return err + } + if err := os.Rename(tmp, path); err != nil { + _ = os.Remove(tmp) + if os.IsPermission(err) { + return fmt.Errorf("config is not writable") + } + return err + } + return nil + } + var transcriber *voice.GroqTranscriber if cfg.Providers.Groq.APIKey != "" { transcriber = voice.NewGroqTranscriber(cfg.Providers.Groq.APIKey) @@ -658,6 +712,37 @@ func gatewayCmd() { fmt.Println("✓ Device event service started") } + shutdown := func(grace time.Duration) { + fmt.Println("\nShutting down...") + + // Stop accepting new inbound work as early as possible. + _ = channelManager.StopAll(context.Background()) + agentLoop.Stop() + + idleCtx, idleCancel := context.WithTimeout(context.Background(), grace) + _ = agentLoop.WaitForIdle(idleCtx) + idleCancel() + + cancel() + deviceService.Stop() + heartbeatService.Stop() + cronService.Stop() + fmt.Println("✓ Gateway stopped") + } + + if webuiCh, ok := channelManager.GetChannel("webui"); ok { + if wc, ok := webuiCh.(*channels.WebUIChannel); ok { + wc.SetConfigUpdate(func(raw []byte) error { + return saveConfigRawAtomic(configPath, raw) + }) + wc.SetDrainExit(func(timeout time.Duration) error { + shutdown(timeout) + os.Exit(0) + return nil + }) + } + } + if err := channelManager.StartAll(ctx); err != nil { fmt.Printf("Error starting channels: %v\n", err) } @@ -667,15 +752,7 @@ func gatewayCmd() { sigChan := make(chan os.Signal, 1) signal.Notify(sigChan, os.Interrupt) <-sigChan - - fmt.Println("\nShutting down...") - cancel() - deviceService.Stop() - heartbeatService.Stop() - cronService.Stop() - agentLoop.Stop() - channelManager.StopAll(ctx) - fmt.Println("✓ Gateway stopped") + shutdown(30 * time.Second) } func statusCmd() { diff --git a/config/config.example.json b/config/config.example.json index 13fa03b9c..6c527735c 100644 --- a/config/config.example.json +++ b/config/config.example.json @@ -128,6 +128,7 @@ "gateway": { "bind": "all", "port": 18790, - "token": "" + "token": "", + "admin_token": "" } } diff --git a/pkg/agent/loop.go b/pkg/agent/loop.go index f3dd94090..c2ed835ed 100644 --- a/pkg/agent/loop.go +++ b/pkg/agent/loop.go @@ -42,6 +42,7 @@ type AgentLoop struct { tools *tools.ToolRegistry running atomic.Bool summarizing sync.Map // Tracks which sessions are currently being summarized + inflight sync.WaitGroup } // processOptions configures how a message is processed @@ -162,7 +163,10 @@ func (al *AgentLoop) Run(ctx context.Context) error { continue } + al.inflight.Add(1) + response, err := al.processMessage(ctx, msg) + al.inflight.Done() if err != nil { response = fmt.Sprintf("Error processing message: %v", err) } @@ -195,6 +199,21 @@ func (al *AgentLoop) Stop() { al.running.Store(false) } +func (al *AgentLoop) WaitForIdle(ctx context.Context) bool { + done := make(chan struct{}) + go func() { + al.inflight.Wait() + close(done) + }() + + select { + case <-done: + return true + case <-ctx.Done(): + return false + } +} + func (al *AgentLoop) RegisterTool(tool tools.Tool) { al.tools.Register(tool) } diff --git a/pkg/channels/webui.go b/pkg/channels/webui.go index 1251ec8c5..2df57826e 100644 --- a/pkg/channels/webui.go +++ b/pkg/channels/webui.go @@ -4,12 +4,14 @@ import ( "context" "encoding/json" "fmt" + "io" "net/http" "net/url" "os" "path/filepath" "strings" "sync" + "sync/atomic" "time" "github.com/gorilla/websocket" @@ -25,6 +27,9 @@ type WebUIChannel struct { httpServer *http.Server mu sync.RWMutex clients map[*webUIClient]struct{} + acceptingWS atomic.Bool + drainExitFn func(timeout time.Duration) error + configUpdateFn func(raw []byte) error } type webUIClient struct { @@ -48,11 +53,21 @@ type webUIOutboundMessage struct { func NewWebUIChannel(cfg config.GatewayConfig, messageBus *bus.MessageBus) (*WebUIChannel, error) { base := NewBaseChannel("webui", cfg, messageBus, nil) - return &WebUIChannel{ + c := &WebUIChannel{ BaseChannel: base, cfg: cfg, clients: make(map[*webUIClient]struct{}), - }, nil + } + c.acceptingWS.Store(true) + return c, nil +} + +func (c *WebUIChannel) SetDrainExit(fn func(timeout time.Duration) error) { + c.drainExitFn = fn +} + +func (c *WebUIChannel) SetConfigUpdate(fn func(raw []byte) error) { + c.configUpdateFn = fn } func (c *WebUIChannel) Start(ctx context.Context) error { @@ -63,6 +78,8 @@ func (c *WebUIChannel) Start(ctx context.Context) error { mux := http.NewServeMux() mux.HandleFunc("/ws", c.handleWS) + mux.HandleFunc("/admin/config", c.handleAdminConfig) + mux.HandleFunc("/admin/drain-exit", c.handleAdminDrainExit) mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) w.Write([]byte("ok")) @@ -91,13 +108,7 @@ func (c *WebUIChannel) Start(ctx context.Context) error { func (c *WebUIChannel) Stop(ctx context.Context) error { c.setRunning(false) - - c.mu.Lock() - for cl := range c.clients { - cl.conn.Close() - delete(c.clients, cl) - } - c.mu.Unlock() + c.beginDrain() if c.httpServer != nil { shutdownCtx, cancel := context.WithTimeout(ctx, 5*time.Second) @@ -161,6 +172,11 @@ func (c *WebUIChannel) isAuthorized(u *url.URL) bool { } func (c *WebUIChannel) handleWS(w http.ResponseWriter, r *http.Request) { + if !c.acceptingWS.Load() { + http.Error(w, "Gateway is draining", http.StatusServiceUnavailable) + return + } + if !c.isAuthorized(r.URL) { http.Error(w, "Unauthorized", http.StatusUnauthorized) return @@ -212,6 +228,9 @@ func (c *WebUIChannel) handleWS(w http.ResponseWriter, r *http.Request) { if err != nil { return } + if !c.acceptingWS.Load() { + return + } var in webUIInboundMessage if err := json.Unmarshal(data, &in); err != nil { @@ -234,10 +253,123 @@ func (c *WebUIChannel) handleWS(w http.ResponseWriter, r *http.Request) { client.chatID = chatID client.sender = senderID + c.bus.PublishInbound(bus.InboundMessage{ + Channel: "webui", + ChatID: chatID, + SenderID: senderID, + Content: content, + SessionKey: chatID, + Metadata: map[string]string{"source": "webui"}, + }) c.HandleMessage(senderID, chatID, content, nil, map[string]string{"source": "webui"}) } } +func (c *WebUIChannel) beginDrain() { + c.acceptingWS.Store(false) + + c.mu.Lock() + for cl := range c.clients { + _ = cl.conn.Close() + delete(c.clients, cl) + } + c.mu.Unlock() +} + +func (c *WebUIChannel) isAdminAuthorized(r *http.Request) bool { + expected := strings.TrimSpace(c.cfg.AdminToken) + if expected == "" { + return false + } + auth := strings.TrimSpace(r.Header.Get("Authorization")) + const prefix = "Bearer " + if !strings.HasPrefix(auth, prefix) { + return false + } + provided := strings.TrimSpace(strings.TrimPrefix(auth, prefix)) + return provided != "" && provided == expected +} + +func (c *WebUIChannel) handleAdminConfig(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPut { + w.Header().Set("Allow", http.MethodPut) + http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) + return + } + if !c.isAdminAuthorized(r) { + http.Error(w, "Unauthorized", http.StatusUnauthorized) + return + } + if c.configUpdateFn == nil { + http.Error(w, "Config update not available", http.StatusNotImplemented) + return + } + body, err := io.ReadAll(io.LimitReader(r.Body, 2<<20)) + if err != nil { + http.Error(w, "Failed to read body", http.StatusBadRequest) + return + } + if len(body) == 0 { + http.Error(w, "Empty body", http.StatusBadRequest) + return + } + if err := c.configUpdateFn(body); err != nil { + msg := err.Error() + lower := strings.ToLower(msg) + if strings.Contains(lower, "config is not writable") { + http.Error(w, msg, http.StatusConflict) + return + } + http.Error(w, msg, http.StatusBadRequest) + return + } + w.WriteHeader(http.StatusOK) + w.Write([]byte("ok")) +} + +type drainExitRequest struct { + TimeoutSeconds int `json:"timeout_seconds"` +} + +func (c *WebUIChannel) handleAdminDrainExit(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + w.Header().Set("Allow", http.MethodPost) + http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) + return + } + if !c.isAdminAuthorized(r) { + http.Error(w, "Unauthorized", http.StatusUnauthorized) + return + } + if c.drainExitFn == nil { + http.Error(w, "Drain/exit not available", http.StatusNotImplemented) + return + } + + // Stop accepting new WS connections and close existing sessions immediately. + c.beginDrain() + + timeout := 30 * time.Second + if r.Body != nil { + body, _ := io.ReadAll(io.LimitReader(r.Body, 64<<10)) + if len(strings.TrimSpace(string(body))) > 0 { + var req drainExitRequest + if err := json.Unmarshal(body, &req); err == nil { + if req.TimeoutSeconds > 0 { + timeout = time.Duration(req.TimeoutSeconds) * time.Second + } + } + } + } + + w.WriteHeader(http.StatusAccepted) + w.Write([]byte("draining")) + + go func() { + _ = c.drainExitFn(timeout) + }() +} + func (c *WebUIChannel) staticHandler() http.Handler { root := c.findUIRoot() fs := http.FileServer(http.Dir(root)) diff --git a/pkg/config/config.go b/pkg/config/config.go index ba5f0b41c..cfcc19d6d 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -199,6 +199,7 @@ type GatewayConfig struct { Bind string `json:"bind" env:"PICOCLAW_GATEWAY_BIND"` Port int `json:"port" env:"PICOCLAW_GATEWAY_PORT"` Token string `json:"token" env:"PICOCLAW_GATEWAY_TOKEN"` + AdminToken string `json:"admin_token" env:"PICOCLAW_GATEWAY_ADMIN_TOKEN"` } type BraveConfig struct { @@ -320,6 +321,7 @@ func DefaultConfig() *Config { Bind: "all", Port: 18790, Token: "", + AdminToken: "", }, Tools: ToolsConfig{ Web: WebToolsConfig{ diff --git a/pkg/migrate/config.go b/pkg/migrate/config.go index fb47bda98..a291f1045 100644 --- a/pkg/migrate/config.go +++ b/pkg/migrate/config.go @@ -214,6 +214,9 @@ func ConvertConfig(data map[string]interface{}) (*config.Config, []string, error if v, ok := getString(gateway, "token"); ok { cfg.Gateway.Token = v } + if v, ok := getString(gateway, "admin_token"); ok { + cfg.Gateway.AdminToken = v + } } if tools, ok := getMap(data, "tools"); ok {