opencode/picoclaw
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Added a remote admin interface that basically can reconfigure the agent

Browse source
This commit is contained in:
Stefan Rinke 2026-02-15 18:20:50 +01:00
parent 6a222045b7
commit 3335ceb9b7
7 changed files with 297 additions and 19 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

44
README.md
View file

@ -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 <admin_token>`
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 <admin_token>" \
-H "Content-Type: application/json" \
--data-binary @config.json \
http://<gateway-host>:18790/admin/config
```
Trigger graceful drain and restart (default timeout 30s):
```bash
curl -X POST \
-H "Authorization: Bearer <admin_token>" \
-H "Content-Type: application/json" \
-d '{"timeout_seconds":30}' \
http://<gateway-host>:18790/admin/drain-exit
```
## CLI Reference
| Command | Description |

95
cmd/picoclaw/main.go
View file

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

3
config/config.example.json
View file

@ -128,6 +128,7 @@
"gateway": {
"bind": "all",
"port": 18790,
"token": ""
"token": "",
"admin_token": ""
}
}

19
pkg/agent/loop.go
View file

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

150
pkg/channels/webui.go
View file

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

2
pkg/config/config.go
View file

@ -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{

3
pkg/migrate/config.go
View file

@ -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 {