From 8f6497987ab151bc83e8772d29297e4612ed5203 Mon Sep 17 00:00:00 2001 From: attaboy11 Date: Sat, 30 May 2026 16:53:30 +0100 Subject: [PATCH] feat: wire edge api to message bus --- README.md | 7 +- cmd/picoclaw/main.go | 4 +- pkg/edge/server.go | 149 ++++++++++++++++++++++++++++++++++++---- pkg/edge/server_test.go | 144 ++++++++++++++++++++++++++++++++++++++ pkg/gene/selector.go | 8 ++- 5 files changed, 292 insertions(+), 20 deletions(-) create mode 100644 pkg/edge/server_test.go diff --git a/README.md b/README.md index 26e1290..8323bcb 100644 --- a/README.md +++ b/README.md @@ -277,9 +277,10 @@ PicoClaw can act as an L1 edge node in a distributed architecture, reporting to ``` When enabled, PicoClaw exposes: -- `GET /healthz` — Health check -- `GET /api/v1/status` — Node status -- `POST /api/v1/command` — Receive commands from fleet +- `GET /api/health` — Health check (`/healthz` remains available) +- `GET /api/status` — Node status (`/api/v1/status` remains available) +- `POST /api/command` — Receive commands from fleet and queue them on MessageBus +- `POST /api/message` — Receive user/operator messages and queue them on MessageBus And periodically sends heartbeats (including gene stats) to the configured fleet manager. diff --git a/cmd/picoclaw/main.go b/cmd/picoclaw/main.go index 269f9b6..67d5b07 100644 --- a/cmd/picoclaw/main.go +++ b/cmd/picoclaw/main.go @@ -588,7 +588,7 @@ func gatewayCmd() { Token: cfg.Edge.CloudToken, }, } - edgeServer = edge.NewServer(edgeCfg) + edgeServer = edge.NewServer(edgeCfg, msgBus) edgeReporter = edge.NewReporter(edgeCfg) // Wire gene engine into edge reporter for heartbeat stats @@ -991,7 +991,7 @@ func cronHelp() { func cronListCmd(storePath string) { cs := cron.NewCronService(storePath, nil) - jobs := cs.ListJobs(true) // Show all jobs, including disabled + jobs := cs.ListJobs(true) // Show all jobs, including disabled if len(jobs) == 0 { fmt.Println("No scheduled jobs.") diff --git a/pkg/edge/server.go b/pkg/edge/server.go index 1896d63..c9c0d0d 100644 --- a/pkg/edge/server.go +++ b/pkg/edge/server.go @@ -6,6 +6,8 @@ import ( "log" "net/http" "time" + + "github.com/sipeed/picoclaw/pkg/bus" ) // Server is the Edge API HTTP server. @@ -15,23 +17,34 @@ type Server struct { config Config mux *http.ServeMux start time.Time + bus *bus.MessageBus } // NewServer creates a new Edge API Server. -func NewServer(cfg Config) *Server { +func NewServer(cfg Config, messageBus ...*bus.MessageBus) *Server { + msgBus := bus.NewMessageBus() + if len(messageBus) > 0 && messageBus[0] != nil { + msgBus = messageBus[0] + } + s := &Server{ config: cfg, mux: http.NewServeMux(), start: time.Now(), + bus: msgBus, } s.routes() return s } func (s *Server) routes() { + s.mux.HandleFunc("GET /api/health", s.handleHealthz) s.mux.HandleFunc("GET /healthz", s.handleHealthz) + s.mux.HandleFunc("GET /api/status", s.handleStatus) s.mux.HandleFunc("GET /api/v1/status", s.handleStatus) + s.mux.HandleFunc("POST /api/command", s.handleCommand) s.mux.HandleFunc("POST /api/v1/command", s.handleCommand) + s.mux.HandleFunc("POST /api/message", s.handleMessage) } // Start begins listening on the configured port. @@ -54,27 +67,135 @@ func (s *Server) handleHealthz(w http.ResponseWriter, r *http.Request) { func (s *Server) handleStatus(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json") json.NewEncoder(w).Encode(map[string]any{ - "node_id": s.config.NodeID, - "node_name": s.config.NodeName, - "status": "running", - "uptime": time.Since(s.start).String(), - "cloud": s.config.Cloud.Endpoint, + "node_id": s.config.NodeID, + "node_name": s.config.NodeName, + "status": "running", + "uptime": time.Since(s.start).String(), + "cloud": s.config.Cloud.Endpoint, + "message_bus": "connected", }) } func (s *Server) handleCommand(w http.ResponseWriter, r *http.Request) { - var cmd struct { - Type string `json:"type"` - Payload map[string]any `json:"payload"` + var req struct { + ID string `json:"id"` + Type string `json:"type"` + Payload map[string]any `json:"payload"` + SenderID string `json:"sender_id"` + ChatID string `json:"chat_id"` + SessionKey string `json:"session_key"` + } + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + http.Error(w, "invalid JSON", http.StatusBadRequest) + return + } + if req.Type == "" { + http.Error(w, "missing command type", http.StatusBadRequest) + return + } + + if req.ID == "" { + req.ID = fmt.Sprintf("cmd-%d", time.Now().UnixNano()) + } + if req.SenderID == "" { + req.SenderID = "fleet" + } + if req.ChatID == "" { + req.ChatID = req.ID + } + if req.SessionKey == "" { + req.SessionKey = "edge-command:" + req.ID + } + + content, err := json.Marshal(map[string]any{ + "id": req.ID, + "type": req.Type, + "payload": req.Payload, + }) + if err != nil { + http.Error(w, "invalid command payload", http.StatusBadRequest) + return + } + + s.bus.PublishInbound(bus.InboundMessage{ + Channel: "edge-api", + SenderID: req.SenderID, + ChatID: req.ChatID, + Content: string(content), + SessionKey: req.SessionKey, + Metadata: map[string]string{ + "kind": "command", + "command_id": req.ID, + "command_type": req.Type, + "source": "edge-http", + }, + }) + + log.Printf("[edge] queued command: id=%s type=%s", req.ID, req.Type) + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusAccepted) + json.NewEncoder(w).Encode(map[string]any{ + "status": "accepted", + "command_id": req.ID, + "command": req.Type, + "queued": true, + }) +} + +func (s *Server) handleMessage(w http.ResponseWriter, r *http.Request) { + var req struct { + Channel string `json:"channel"` + SenderID string `json:"sender_id"` + ChatID string `json:"chat_id"` + Content string `json:"content"` + Media []string `json:"media,omitempty"` + SessionKey string `json:"session_key"` + Metadata map[string]string `json:"metadata,omitempty"` } - if err := json.NewDecoder(r.Body).Decode(&cmd); err != nil { + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { http.Error(w, "invalid JSON", http.StatusBadRequest) return } - log.Printf("[edge] received command: type=%s", cmd.Type) + if req.Content == "" { + http.Error(w, "missing message content", http.StatusBadRequest) + return + } + + if req.Channel == "" { + req.Channel = "edge-api" + } + if req.SenderID == "" { + req.SenderID = "fleet" + } + if req.ChatID == "" { + req.ChatID = s.config.NodeID + } + if req.SessionKey == "" { + req.SessionKey = "edge-message:" + req.ChatID + } + if req.Metadata == nil { + req.Metadata = make(map[string]string) + } + req.Metadata["kind"] = "message" + req.Metadata["source"] = "edge-http" + + s.bus.PublishInbound(bus.InboundMessage{ + Channel: req.Channel, + SenderID: req.SenderID, + ChatID: req.ChatID, + Content: req.Content, + Media: req.Media, + SessionKey: req.SessionKey, + Metadata: req.Metadata, + }) + + log.Printf("[edge] queued message: channel=%s sender=%s", req.Channel, req.SenderID) w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusAccepted) json.NewEncoder(w).Encode(map[string]any{ - "status": "accepted", - "command": cmd.Type, + "status": "accepted", + "channel": req.Channel, + "session_key": req.SessionKey, + "queued": true, }) -} \ No newline at end of file +} diff --git a/pkg/edge/server_test.go b/pkg/edge/server_test.go new file mode 100644 index 0000000..ec0f790 --- /dev/null +++ b/pkg/edge/server_test.go @@ -0,0 +1,144 @@ +package edge + +import ( + "bytes" + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/sipeed/picoclaw/pkg/bus" +) + +func testServer() (*Server, *bus.MessageBus) { + msgBus := bus.NewMessageBus() + cfg := Config{ + Port: 9090, + NodeID: "edge-01", + NodeName: "Edge 01", + Cloud: CloudConfig{ + Endpoint: "http://fleet.local", + }, + } + return NewServer(cfg, msgBus), msgBus +} + +func decodeJSON(t *testing.T, rr *httptest.ResponseRecorder) map[string]any { + t.Helper() + var body map[string]any + if err := json.NewDecoder(rr.Body).Decode(&body); err != nil { + t.Fatalf("decode response JSON: %v", err) + } + return body +} + +func TestHealthEndpoint(t *testing.T) { + s, _ := testServer() + req := httptest.NewRequest(http.MethodGet, "/api/health", nil) + rr := httptest.NewRecorder() + + s.mux.ServeHTTP(rr, req) + + if rr.Code != http.StatusOK { + t.Fatalf("expected 200, got %d", rr.Code) + } + body := decodeJSON(t, rr) + if body["status"] != "ok" { + t.Fatalf("expected ok status, got %#v", body["status"]) + } + if body["node_id"] != "edge-01" { + t.Fatalf("expected node id in health response, got %#v", body["node_id"]) + } +} + +func TestStatusEndpoint(t *testing.T) { + s, _ := testServer() + req := httptest.NewRequest(http.MethodGet, "/api/status", nil) + rr := httptest.NewRecorder() + + s.mux.ServeHTTP(rr, req) + + if rr.Code != http.StatusOK { + t.Fatalf("expected 200, got %d", rr.Code) + } + body := decodeJSON(t, rr) + if body["node_name"] != "Edge 01" { + t.Fatalf("expected node name in status response, got %#v", body["node_name"]) + } + if body["message_bus"] != "connected" { + t.Fatalf("expected message bus status, got %#v", body["message_bus"]) + } +} + +func TestCommandEndpointQueuesMessageBusCommand(t *testing.T) { + s, msgBus := testServer() + payload := []byte(`{"id":"cmd-1","type":"restart","payload":{"delay_seconds":5}}`) + req := httptest.NewRequest(http.MethodPost, "/api/command", bytes.NewReader(payload)) + rr := httptest.NewRecorder() + + s.mux.ServeHTTP(rr, req) + + if rr.Code != http.StatusAccepted { + t.Fatalf("expected 202, got %d: %s", rr.Code, rr.Body.String()) + } + body := decodeJSON(t, rr) + if body["queued"] != true { + t.Fatalf("expected queued response, got %#v", body["queued"]) + } + + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + msg, ok := msgBus.ConsumeInbound(ctx) + if !ok { + t.Fatal("expected command on message bus") + } + if msg.Channel != "edge-api" { + t.Fatalf("expected edge-api channel, got %s", msg.Channel) + } + if msg.Metadata["kind"] != "command" || msg.Metadata["command_type"] != "restart" { + t.Fatalf("unexpected command metadata: %#v", msg.Metadata) + } + if msg.Content == "" { + t.Fatal("expected command content") + } +} + +func TestMessageEndpointQueuesInboundMessage(t *testing.T) { + s, msgBus := testServer() + payload := []byte(`{"content":"check pond oxygen","sender_id":"operator","chat_id":"pond-01","metadata":{"priority":"high"}}`) + req := httptest.NewRequest(http.MethodPost, "/api/message", bytes.NewReader(payload)) + rr := httptest.NewRecorder() + + s.mux.ServeHTTP(rr, req) + + if rr.Code != http.StatusAccepted { + t.Fatalf("expected 202, got %d: %s", rr.Code, rr.Body.String()) + } + + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + msg, ok := msgBus.ConsumeInbound(ctx) + if !ok { + t.Fatal("expected inbound message on message bus") + } + if msg.Content != "check pond oxygen" { + t.Fatalf("unexpected message content: %s", msg.Content) + } + if msg.Metadata["priority"] != "high" || msg.Metadata["source"] != "edge-http" { + t.Fatalf("unexpected metadata: %#v", msg.Metadata) + } +} + +func TestCommandEndpointRejectsMissingType(t *testing.T) { + s, _ := testServer() + req := httptest.NewRequest(http.MethodPost, "/api/command", bytes.NewReader([]byte(`{"payload":{}}`))) + rr := httptest.NewRecorder() + + s.mux.ServeHTTP(rr, req) + + if rr.Code != http.StatusBadRequest { + t.Fatalf("expected 400, got %d", rr.Code) + } +} diff --git a/pkg/gene/selector.go b/pkg/gene/selector.go index 40e31ef..fa14fe2 100644 --- a/pkg/gene/selector.go +++ b/pkg/gene/selector.go @@ -83,7 +83,13 @@ func SelectGenes(genes []Gene, signals []string, preset StrategyPreset, maxResul // Sort by score descending sort.Slice(candidates, func(i, j int) bool { - return candidates[i].Score > candidates[j].Score + if candidates[i].Score != candidates[j].Score { + return candidates[i].Score > candidates[j].Score + } + if candidates[i].Gene.Confidence != candidates[j].Gene.Confidence { + return candidates[i].Gene.Confidence > candidates[j].Gene.Confidence + } + return candidates[i].Gene.VerifiedBy > candidates[j].Gene.VerifiedBy }) if maxResults > 0 && len(candidates) > maxResults {