feat: wire edge API to message bus

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.
 

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.")

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,
 	})
-}
+}

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

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 {