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..9abc290 100644 --- a/pkg/edge/server.go +++ b/pkg/edge/server.go @@ -5,7 +5,10 @@ import ( "fmt" "log" "net/http" + "strings" "time" + + "github.com/sipeed/picoclaw/pkg/bus" ) // Server is the Edge API HTTP server. @@ -15,23 +18,44 @@ 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 { + var msgBus *bus.MessageBus + if len(messageBus) > 0 { + 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 /healthz", s.handleHealthz) - s.mux.HandleFunc("GET /api/v1/status", s.handleStatus) - s.mux.HandleFunc("POST /api/v1/command", s.handleCommand) + for _, route := range []string{"GET /healthz", "GET /api/health", "GET /api/v1/health"} { + s.mux.HandleFunc(route, s.handleHealthz) + } + for _, route := range []string{"GET /api/status", "GET /api/v1/status"} { + s.mux.HandleFunc(route, s.handleStatus) + } + for _, route := range []string{"POST /api/command", "POST /api/v1/command"} { + s.mux.HandleFunc(route, s.handleCommand) + } + for _, route := range []string{"POST /api/message", "POST /api/v1/message"} { + s.mux.HandleFunc(route, s.handleMessage) + } +} + +// ServeHTTP lets callers embed or test the Edge API server as a normal HTTP handler. +func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) { + s.mux.ServeHTTP(w, r) } // Start begins listening on the configured port. @@ -42,39 +66,191 @@ func (s *Server) Start() error { } func (s *Server) handleHealthz(w http.ResponseWriter, r *http.Request) { - w.Header().Set("Content-Type", "application/json") - json.NewEncoder(w).Encode(map[string]any{ - "status": "ok", - "agent": "picclaw", - "node_id": s.config.NodeID, - "uptime": time.Since(s.start).String(), + writeJSON(w, http.StatusOK, map[string]any{ + "status": "ok", + "agent": "picclaw", + "node_id": s.config.NodeID, + "node_name": s.config.NodeName, + "uptime": time.Since(s.start).String(), + "bus_connected": s.bus != nil, }) } 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, + writeJSON(w, http.StatusOK, 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, + "bus_connected": s.bus != nil, + "endpoints": []string{ + "GET /healthz", + "GET /api/health", + "GET /api/status", + "POST /api/command", + "POST /api/message", + }, }) } 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 edgeCommandRequest + if err := decodeJSON(r, &req); err != nil { + writeError(w, http.StatusBadRequest, "invalid JSON") + return } - if err := json.NewDecoder(r.Body).Decode(&cmd); err != nil { - http.Error(w, "invalid JSON", http.StatusBadRequest) + req.CommandID = firstNonEmpty(req.CommandID, req.ID, fmt.Sprintf("cmd-%d", time.Now().UnixNano())) + req.Type = strings.TrimSpace(req.Type) + if req.Type == "" { + writeError(w, http.StatusBadRequest, "command type is required") return } - log.Printf("[edge] received command: type=%s", cmd.Type) - w.Header().Set("Content-Type", "application/json") - json.NewEncoder(w).Encode(map[string]any{ - "status": "accepted", - "command": cmd.Type, + + if s.bus != nil { + s.bus.PublishInbound(req.toInbound(s.config.NodeID)) + } + + log.Printf("[edge] queued command: id=%s type=%s", req.CommandID, req.Type) + writeJSON(w, http.StatusAccepted, map[string]any{ + "status": "accepted", + "command_id": req.CommandID, + "command": req.Type, + "queued": s.bus != nil, + }) +} + +func (s *Server) handleMessage(w http.ResponseWriter, r *http.Request) { + var req edgeMessageRequest + if err := decodeJSON(r, &req); err != nil { + writeError(w, http.StatusBadRequest, "invalid JSON") + return + } + req.MessageID = firstNonEmpty(req.MessageID, req.ID, fmt.Sprintf("msg-%d", time.Now().UnixNano())) + req.Content = strings.TrimSpace(req.Content) + if req.Content == "" { + writeError(w, http.StatusBadRequest, "message content is required") + return + } + + if s.bus != nil { + s.bus.PublishInbound(req.toInbound(s.config.NodeID)) + } + + log.Printf("[edge] queued message: id=%s sender=%s", req.MessageID, firstNonEmpty(req.SenderID, "fleet")) + writeJSON(w, http.StatusAccepted, map[string]any{ + "status": "accepted", + "message_id": req.MessageID, + "queued": s.bus != nil, + }) +} + +type edgeCommandRequest struct { + ID string `json:"id"` + CommandID string `json:"command_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"` + Metadata map[string]string `json:"metadata"` +} + +type edgeMessageRequest struct { + ID string `json:"id"` + MessageID string `json:"message_id"` + 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"` +} + +func (req edgeCommandRequest) toInbound(nodeID string) bus.InboundMessage { + metadata := cloneMetadata(req.Metadata) + metadata["kind"] = "command" + metadata["source"] = "edge-http" + metadata["edge_node_id"] = nodeID + metadata["command_id"] = req.CommandID + metadata["command_type"] = req.Type + metadata["payload"] = jsonString(req.Payload) + + content := jsonString(map[string]any{ + "id": req.CommandID, + "type": req.Type, + "payload": req.Payload, }) -} \ No newline at end of file + + return bus.InboundMessage{ + Channel: "edge-api", + SenderID: firstNonEmpty(req.SenderID, "fleet"), + ChatID: firstNonEmpty(req.ChatID, req.CommandID, nodeID), + Content: content, + SessionKey: firstNonEmpty(req.SessionKey, "edge-command:"+req.CommandID), + Metadata: metadata, + } +} + +func (req edgeMessageRequest) toInbound(nodeID string) bus.InboundMessage { + metadata := cloneMetadata(req.Metadata) + metadata["kind"] = "message" + metadata["source"] = "edge-http" + metadata["edge_node_id"] = nodeID + metadata["message_id"] = req.MessageID + + return bus.InboundMessage{ + Channel: firstNonEmpty(req.Channel, "edge-api"), + SenderID: firstNonEmpty(req.SenderID, "fleet"), + ChatID: firstNonEmpty(req.ChatID, req.MessageID, nodeID), + Content: req.Content, + Media: req.Media, + SessionKey: firstNonEmpty(req.SessionKey, "edge-message:"+firstNonEmpty(req.ChatID, req.MessageID, nodeID)), + Metadata: metadata, + } +} + +func decodeJSON(r *http.Request, target any) error { + decoder := json.NewDecoder(r.Body) + decoder.DisallowUnknownFields() + return decoder.Decode(target) +} + +func writeJSON(w http.ResponseWriter, status int, payload any) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + json.NewEncoder(w).Encode(payload) +} + +func writeError(w http.ResponseWriter, status int, message string) { + writeJSON(w, status, map[string]string{"error": message}) +} + +func cloneMetadata(metadata map[string]string) map[string]string { + out := make(map[string]string, len(metadata)+5) + for key, value := range metadata { + out[key] = value + } + return out +} + +func jsonString(value any) string { + if value == nil { + return "{}" + } + encoded, err := json.Marshal(value) + if err != nil { + return "{}" + } + return string(encoded) +} + +func firstNonEmpty(values ...string) string { + for _, value := range values { + if value != "" { + return value + } + } + return "" +} diff --git a/pkg/edge/server_test.go b/pkg/edge/server_test.go new file mode 100644 index 0000000..1eda7be --- /dev/null +++ b/pkg/edge/server_test.go @@ -0,0 +1,136 @@ +package edge + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/sipeed/picoclaw/pkg/bus" +) + +func newTestServer(messageBus *bus.MessageBus) *Server { + return NewServer(Config{NodeID: "edge-01", NodeName: "PicClaw Lab", Cloud: CloudConfig{Endpoint: "https://fleet.example.test"}}, messageBus) +} + +func TestHealthAndStatusRoutes(t *testing.T) { + srv := newTestServer(bus.NewMessageBus()) + for _, path := range []string{"/healthz", "/api/health", "/api/v1/health"} { + rec := httptest.NewRecorder() + srv.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, path, nil)) + if rec.Code != http.StatusOK { + t.Fatalf("%s returned %d", path, rec.Code) + } + var body map[string]any + if err := json.Unmarshal(rec.Body.Bytes(), &body); err != nil { + t.Fatal(err) + } + if body["status"] != "ok" || body["node_id"] != "edge-01" || body["bus_connected"] != true { + t.Fatalf("bad health body for %s: %#v", path, body) + } + } + + rec := httptest.NewRecorder() + srv.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/api/status", nil)) + if rec.Code != http.StatusOK { + t.Fatalf("status returned %d", rec.Code) + } + var status map[string]any + if err := json.Unmarshal(rec.Body.Bytes(), &status); err != nil { + t.Fatal(err) + } + if status["cloud"] != "https://fleet.example.test" || status["status"] != "running" { + t.Fatalf("bad status: %#v", status) + } +} + +func TestCommandRoutesPublishToMessageBus(t *testing.T) { + for _, path := range []string{"/api/command", "/api/v1/command"} { + msgBus := bus.NewMessageBus() + srv := newTestServer(msgBus) + rec := httptest.NewRecorder() + srv.ServeHTTP(rec, httptest.NewRequest(http.MethodPost, path, strings.NewReader(`{"id":"legacy-id","command_id":"cmd-1","type":"restart_sensor","payload":{"sensor":"temp"},"sender_id":"fleet","session_key":"session-1"}`))) + if rec.Code != http.StatusAccepted { + t.Fatalf("%s returned %d: %s", path, rec.Code, rec.Body.String()) + } + msg := consumeInbound(t, msgBus) + if msg.Channel != "edge-api" || msg.ChatID != "cmd-1" || msg.SessionKey != "session-1" { + t.Fatalf("bad command message: %#v", msg) + } + if msg.Metadata["kind"] != "command" || msg.Metadata["command_type"] != "restart_sensor" || msg.Metadata["payload"] != `{"sensor":"temp"}` { + t.Fatalf("bad command metadata: %#v", msg.Metadata) + } + if !strings.Contains(msg.Content, "restart_sensor") { + t.Fatalf("content should include command details: %s", msg.Content) + } + } +} + +func TestMessageRoutesPublishToMessageBus(t *testing.T) { + for _, path := range []string{"/api/message", "/api/v1/message"} { + msgBus := bus.NewMessageBus() + srv := newTestServer(msgBus) + rec := httptest.NewRecorder() + srv.ServeHTTP(rec, httptest.NewRequest(http.MethodPost, path, strings.NewReader(`{"message_id":"msg-1","channel":"ops","sender_id":"fleet","chat_id":"pond","content":"check oxygen","media":["image.jpg"],"metadata":{"priority":"high"}}`))) + if rec.Code != http.StatusAccepted { + t.Fatalf("%s returned %d: %s", path, rec.Code, rec.Body.String()) + } + msg := consumeInbound(t, msgBus) + if msg.Channel != "ops" || msg.Content != "check oxygen" || msg.ChatID != "pond" { + t.Fatalf("bad message: %#v", msg) + } + if len(msg.Media) != 1 || msg.Media[0] != "image.jpg" { + t.Fatalf("bad media: %#v", msg.Media) + } + if msg.Metadata["priority"] != "high" || msg.Metadata["kind"] != "message" || msg.Metadata["source"] != "edge-http" { + t.Fatalf("bad message metadata: %#v", msg.Metadata) + } + } +} + +func TestInvalidRequestsReturnBadRequest(t *testing.T) { + srv := newTestServer(bus.NewMessageBus()) + cases := []struct{ method, path, body string }{ + {http.MethodPost, "/api/command", "{"}, + {http.MethodPost, "/api/command", `{"payload":{}}`}, + {http.MethodPost, "/api/message", "{"}, + {http.MethodPost, "/api/message", `{"content":""}`}, + } + for _, tc := range cases { + rec := httptest.NewRecorder() + srv.ServeHTTP(rec, httptest.NewRequest(tc.method, tc.path, strings.NewReader(tc.body))) + if rec.Code != http.StatusBadRequest { + t.Fatalf("%s %s returned %d", tc.method, tc.path, rec.Code) + } + } +} + +func TestServerWithoutMessageBusStillAccepts(t *testing.T) { + srv := newTestServer(nil) + rec := httptest.NewRecorder() + srv.ServeHTTP(rec, httptest.NewRequest(http.MethodPost, "/api/command", strings.NewReader(`{"type":"status"}`))) + if rec.Code != http.StatusAccepted { + t.Fatalf("expected accepted without bus, got %d", rec.Code) + } + var body map[string]any + if err := json.Unmarshal(rec.Body.Bytes(), &body); err != nil { + t.Fatal(err) + } + if body["queued"] != false { + t.Fatalf("expected queued=false without bus: %#v", body) + } +} + +func consumeInbound(t *testing.T, msgBus *bus.MessageBus) bus.InboundMessage { + t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + msg, ok := msgBus.ConsumeInbound(ctx) + if !ok { + t.Fatal("expected inbound message") + } + return msg +} diff --git a/pkg/gene/selector.go b/pkg/gene/selector.go index 40e31ef..8ec7d6f 100644 --- a/pkg/gene/selector.go +++ b/pkg/gene/selector.go @@ -81,9 +81,20 @@ func SelectGenes(genes []Gene, signals []string, preset StrategyPreset, maxResul } } - // Sort by score descending + // Sort by score descending. When scores are capped/tied, keep the + // ordering deterministic by preferring stronger confidence and + // verification signals before falling back to ID. 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 + } + if candidates[i].Gene.VerifiedBy != candidates[j].Gene.VerifiedBy { + return candidates[i].Gene.VerifiedBy > candidates[j].Gene.VerifiedBy + } + return candidates[i].Gene.ID < candidates[j].Gene.ID }) if maxResults > 0 && len(candidates) > maxResults {