diff --git a/.golangci.yaml b/.golangci.yaml index b2b772406..3527a4251 100644 --- a/.golangci.yaml +++ b/.golangci.yaml @@ -1,5 +1,3 @@ -version: "2" - linters: default: all disable: diff --git a/Makefile b/Makefile index 4704b7c4a..00236ff6e 100644 --- a/Makefile +++ b/Makefile @@ -273,7 +273,7 @@ test: generate ## fmt: Format Go code fmt: - @$(GOLANGCI_LINT) fmt + @$(GO) fmt ./... ## lint: Run linters lint: diff --git a/pkg/channels/http/http.go b/pkg/channels/http/http.go new file mode 100644 index 000000000..26470f6d8 --- /dev/null +++ b/pkg/channels/http/http.go @@ -0,0 +1,45 @@ +package http + +import ( + "context" + + "github.com/sipeed/picoclaw/pkg/bus" + "github.com/sipeed/picoclaw/pkg/channels" + "github.com/sipeed/picoclaw/pkg/config" + "github.com/sipeed/picoclaw/pkg/logger" +) + +func init() { + channels.RegisterFactory("http", NewHTTPChannel) +} + +type HTTPChannel struct { + *channels.BaseChannel +} + +func NewHTTPChannel(cfg *config.Config, b *bus.MessageBus) (channels.Channel, error) { + bc := channels.NewBaseChannel("http", nil, b, nil) + return &HTTPChannel{ + BaseChannel: bc, + }, nil +} + +func (c *HTTPChannel) Start(ctx context.Context) error { + c.SetRunning(true) + return nil +} + +func (c *HTTPChannel) Stop(ctx context.Context) error { + c.SetRunning(false) + return nil +} + +func (c *HTTPChannel) Send(ctx context.Context, msg bus.OutboundMessage) ([]string, error) { + logger.InfoCF("channels", "HTTP channel received outbound message", map[string]any{ + "chat_id": msg.ChatID, + "content": msg.Content, + }) + // For synchronous HTTP, the response is usually handled by the caller of ProcessDirectWithChannel. + // Asynchronous messages (e.g. from subagents) will just be logged here for now. + return nil, nil +} diff --git a/pkg/config/gateway.go b/pkg/config/gateway.go index e9f4085d3..06df7e5bb 100644 --- a/pkg/config/gateway.go +++ b/pkg/config/gateway.go @@ -10,10 +10,12 @@ import ( const DefaultGatewayLogLevel = "warn" type GatewayConfig struct { - Host string `json:"host" env:"PICOCLAW_GATEWAY_HOST"` - Port int `json:"port" env:"PICOCLAW_GATEWAY_PORT"` - HotReload bool `json:"hot_reload" env:"PICOCLAW_GATEWAY_HOT_RELOAD"` - LogLevel string `json:"log_level,omitempty" env:"PICOCLAW_LOG_LEVEL"` + Host string `json:"host" env:"PICOCLAW_GATEWAY_HOST"` + Port int `json:"port" env:"PICOCLAW_GATEWAY_PORT"` + APIKey string `json:"api_key" env:"PICOCLAW_GATEWAY_API_KEY"` + ChatEnabled bool `json:"chat_enabled" env:"PICOCLAW_GATEWAY_CHAT_ENABLED"` + HotReload bool `json:"hot_reload" env:"PICOCLAW_GATEWAY_HOT_RELOAD"` + LogLevel string `json:"log_level,omitempty" env:"PICOCLAW_LOG_LEVEL"` } func canonicalGatewayLogLevel(level logger.LogLevel) string { diff --git a/pkg/gateway/gateway.go b/pkg/gateway/gateway.go index 509b5d37e..bd22568ae 100644 --- a/pkg/gateway/gateway.go +++ b/pkg/gateway/gateway.go @@ -21,6 +21,7 @@ import ( _ "github.com/sipeed/picoclaw/pkg/channels/dingtalk" _ "github.com/sipeed/picoclaw/pkg/channels/discord" _ "github.com/sipeed/picoclaw/pkg/channels/feishu" + _ "github.com/sipeed/picoclaw/pkg/channels/http" _ "github.com/sipeed/picoclaw/pkg/channels/irc" _ "github.com/sipeed/picoclaw/pkg/channels/line" _ "github.com/sipeed/picoclaw/pkg/channels/maixcam" @@ -111,27 +112,39 @@ func (p *startupBlockedProvider) GetDefaultModel() string { // Run starts the gateway runtime using the configuration loaded from configPath. func Run(debug bool, homePath, configPath string, allowEmptyStartup bool) error { + fmt.Printf("🚀 PicoClaw Gateway starting...\n") + fmt.Printf("📂 Home Path: %s\n", homePath) + fmt.Printf("📄 Config Path: %s\n", configPath) + panicPath := filepath.Join(homePath, logPath, panicFile) + fmt.Printf("🔧 Initializing panic log: %s\n", panicPath) panicFunc, err := logger.InitPanic(panicPath) if err != nil { - return fmt.Errorf("error initializing panic log: %w", err) + fmt.Printf("⚠️ Warning: error initializing panic log (continuing): %v\n", err) + } else if panicFunc != nil { + defer panicFunc() + fmt.Println("✓ Panic log initialized") } - defer panicFunc() - if err = logger.EnableFileLogging(filepath.Join(homePath, logPath, logFile)); err != nil { - logger.Fatal(fmt.Sprintf("error enabling file logging: %v", err)) + logFilePath := filepath.Join(homePath, logPath, logFile) + fmt.Printf("🔧 Enabling file logging: %s\n", logFilePath) + if err = logger.EnableFileLogging(logFilePath); err != nil { + fmt.Printf("⚠️ Warning: error enabling file logging (continuing): %v\n", err) + } else { + defer logger.DisableFileLogging() + fmt.Println("✓ File logging enabled") + } + + fmt.Println("🔍 Loading configuration...") + cfg, err := config.LoadConfig(configPath) + if err != nil { + return fmt.Errorf("error loading config: %w", err) } - defer logger.DisableFileLogging() if debug { logger.SetLevel(logger.DEBUG) } else { - logger.SetLevelFromString(config.ResolveGatewayLogLevel(configPath)) - } - - cfg, err := config.LoadConfig(configPath) - if err != nil { - logger.Fatalf("error loading config: %v", err) + logger.SetLevelFromString(cfg.Gateway.LogLevel) } if err = preCheckConfig(cfg); err != nil { @@ -155,10 +168,14 @@ func Run(debug bool, homePath, configPath string, allowEmptyStartup bool) error } defer pid.RemovePidFile(homePath) + fmt.Printf("🔍 Creating startup provider for model: %s (allow empty: %v)\n", + cfg.Agents.Defaults.GetModelName(), allowEmptyStartup) provider, modelID, err := createStartupProvider(cfg, allowEmptyStartup) if err != nil { + fmt.Printf("❌ Error creating provider: %v\n", err) return fmt.Errorf("error creating provider: %w", err) } + fmt.Printf("✓ Provider created (Model ID: %s)\n", modelID) if modelID != "" { cfg.Agents.Defaults.ModelName = modelID @@ -181,8 +198,10 @@ func Run(debug bool, homePath, configPath string, allowEmptyStartup bool) error "skills_available": skillsInfo["available"], }) + fmt.Println("🚀 Setting up services...") runningServices, err := setupAndStartServices(cfg, agentLoop, msgBus, pidData.Token) if err != nil { + fmt.Printf("❌ Error starting services: %v\n", err) return err } @@ -203,8 +222,23 @@ func Run(debug bool, homePath, configPath string, allowEmptyStartup bool) error } } runningServices.HealthServer.SetReloadFunc(reloadTrigger) + runningServices.HealthServer.SetAPIKey(cfg.Gateway.APIKey) agentLoop.SetReloadFunc(reloadTrigger) + // Setup synchronous /chat endpoint handler + if cfg.Gateway.ChatEnabled { + runningServices.HealthServer.SetChatFunc(func(ctx context.Context, message, sessionID, chatID string) (string, error) { + if sessionID == "" { + sessionID = fmt.Sprintf("chat-%s", time.Now().Format("20060102-150405")) + } + if chatID == "" { + // Default to sessionID to ensure isolation + chatID = sessionID + } + return agentLoop.ProcessDirectWithChannel(ctx, message, sessionID, "http", chatID) + }) + } + fmt.Printf("✓ Gateway started on %s:%d\n", cfg.Gateway.Host, cfg.Gateway.Port) fmt.Println("Press Ctrl+C to stop") diff --git a/pkg/health/server.go b/pkg/health/server.go index 2602cb965..1d616b07b 100644 --- a/pkg/health/server.go +++ b/pkg/health/server.go @@ -7,18 +7,54 @@ import ( "fmt" "maps" "net/http" + "os" + "strings" "sync" "time" + + "github.com/sipeed/picoclaw/pkg/logger" ) +// Mux defines the interface required for registering health handlers. +type Mux interface { + HandleFunc(pattern string, handler func(http.ResponseWriter, *http.Request)) +} + +// ChatRequest is the JSON body for POST /chat. +type ChatRequest struct { + Message string `json:"message"` + SessionID string `json:"session_id,omitempty"` + ChatID string `json:"chat_id,omitempty"` // Alias for session_id to match PicoClaw terminology +} + +// ChatResponse is the JSON response from /chat. +type ChatResponse struct { + Response string `json:"response,omitempty"` + SessionID string `json:"session_id,omitempty"` + Status string `json:"status,omitempty"` + Error string `json:"error,omitempty"` +} + +type chatStatus struct { + Response string + Error error + Done bool + CreatedAt time.Time +} + type Server struct { - server *http.Server - mu sync.RWMutex - ready bool - checks map[string]Check - startTime time.Time - reloadFunc func() error - authToken string // optional bearer token for protected endpoints + server *http.Server + mu sync.RWMutex + ready bool + checks map[string]Check + startTime time.Time + reloadFunc func() error + authToken string // optional bearer token for protected endpoints + chatFunc func(ctx context.Context, message, sessionID, chatID string) (string, error) + apiKey string + chatResults map[string]*chatStatus + chatResultsMu sync.RWMutex + rateLimits sync.Map // key: string (ID or IP), value: time.Time } type Check struct { @@ -32,27 +68,34 @@ type StatusResponse struct { Status string `json:"status"` Uptime string `json:"uptime"` Checks map[string]Check `json:"checks,omitempty"` + Pid int `json:"pid"` } func NewServer(host string, port int, token string) *Server { mux := http.NewServeMux() s := &Server{ - ready: false, - checks: make(map[string]Check), - startTime: time.Now(), - authToken: token, + ready: false, + checks: make(map[string]Check), + startTime: time.Now(), + authToken: token, + chatResults: make(map[string]*chatStatus), } mux.HandleFunc("/health", s.healthHandler) mux.HandleFunc("/ready", s.readyHandler) mux.HandleFunc("/reload", s.reloadHandler) + mux.HandleFunc("/chat", s.chatHandler) + + // Start task cleanup goroutine + go s.taskCleanupLoop() addr := fmt.Sprintf("%s:%d", host, port) s.server = &http.Server{ - Addr: addr, - Handler: mux, - ReadTimeout: 5 * time.Second, - WriteTimeout: 5 * time.Second, + Addr: addr, + Handler: mux, + ReadTimeout: 10 * time.Second, + // WriteTimeout must be long enough for LLM inference; 5 min is generous. + WriteTimeout: 5 * time.Minute, } return s @@ -116,6 +159,62 @@ func (s *Server) SetReloadFunc(fn func() error) { s.reloadFunc = fn } +// SetChatFunc sets the callback that processes /chat requests. +// fn receives the user message and an optional session ID and must return the +// agent's reply (or an error). It is called synchronously inside the HTTP +// handler, so the write timeout on the server governs the maximum duration. +func (s *Server) SetChatFunc(fn func(ctx context.Context, message, sessionID, chatID string) (string, error)) { + s.mu.Lock() + defer s.mu.Unlock() + s.chatFunc = fn +} + +// SetAPIKey sets the expected X-API-Key header value. +func (s *Server) SetAPIKey(key string) { + s.mu.Lock() + defer s.mu.Unlock() + s.apiKey = key +} + +// SetAuthToken sets the expected Bearer token. +func (s *Server) SetAuthToken(token string) { + s.mu.Lock() + defer s.mu.Unlock() + s.authToken = token +} + +func (s *Server) verifyAuth(r *http.Request) bool { + s.mu.RLock() + defer s.mu.RUnlock() + + // If no authentication is configured, allow the request. + if s.apiKey == "" && s.authToken == "" { + return true + } + + // Check X-API-Key header. + if s.apiKey != "" { + gotKey := r.Header.Get("X-API-Key") + if subtle.ConstantTimeCompare([]byte(gotKey), []byte(s.apiKey)) == 1 { + return true + } + } + + // Check Authorization: Bearer header. + if s.authToken != "" { + authHeader := r.Header.Get("Authorization") + const prefix = "Bearer " + if len(authHeader) > len(prefix) && strings.EqualFold(authHeader[:len(prefix)], prefix) { + gotToken := authHeader[len(prefix):] + if subtle.ConstantTimeCompare([]byte(gotToken), []byte(s.authToken)) == 1 { + return true + } + } + } + + return false +} + func (s *Server) reloadHandler(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost { w.Header().Set("Content-Type", "application/json") @@ -123,20 +222,11 @@ func (s *Server) reloadHandler(w http.ResponseWriter, r *http.Request) { json.NewEncoder(w).Encode(map[string]string{"error": "method not allowed, use POST"}) return } - - // Token check - s.mu.RLock() - requiredToken := s.authToken - s.mu.RUnlock() - - if requiredToken != "" { - given := extractBearerToken(r.Header.Get("Authorization")) - if given == "" || subtle.ConstantTimeCompare([]byte(given), []byte(requiredToken)) != 1 { - w.Header().Set("Content-Type", "application/json") - w.WriteHeader(http.StatusUnauthorized) - json.NewEncoder(w).Encode(map[string]string{"error": "unauthorized"}) - return - } + if !s.verifyAuth(r) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusUnauthorized) + json.NewEncoder(w).Encode(map[string]string{"error": "unauthorized"}) + return } s.mu.Lock() @@ -170,6 +260,7 @@ func (s *Server) healthHandler(w http.ResponseWriter, r *http.Request) { resp := StatusResponse{ Status: "ok", Uptime: uptime.String(), + Pid: os.Getpid(), } json.NewEncoder(w).Encode(resp) @@ -213,20 +304,284 @@ func (s *Server) readyHandler(w http.ResponseWriter, r *http.Request) { }) } -// HandlerMux is the interface for registering HTTP handlers, used by -// RegisterOnMux so that callers can pass any mux implementation -// (e.g. *http.ServeMux or a custom dynamic mux). -type HandlerMux interface { - Handle(pattern string, handler http.Handler) - HandleFunc(pattern string, handler func(http.ResponseWriter, *http.Request)) -} - -// RegisterOnMux registers /health, /ready and /reload handlers onto the given mux. -// This allows the health endpoints to be served by a shared HTTP server. -func (s *Server) RegisterOnMux(mux HandlerMux) { +// RegisterOnMux registers /health, /ready, /reload and /chat handlers onto the +// given mux. This allows the health endpoints to be served by a shared HTTP server. +func (s *Server) RegisterOnMux(mux Mux) { mux.HandleFunc("/health", s.healthHandler) mux.HandleFunc("/ready", s.readyHandler) mux.HandleFunc("/reload", s.reloadHandler) + mux.HandleFunc("/chat", s.chatHandler) +} + +// chatHandler handles POST /chat (initiate async) and GET /chat (poll for result). +// POST body: {"message": "...", "session_id": "..." (optional)} +// POST response: {"session_id": "...", "status": "pending"} +// GET query: ?session_id=... +// GET response: {"response": "...", "status": "completed"} +func (s *Server) chatHandler(w http.ResponseWriter, r *http.Request) { + if !s.verifyAuth(r) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusUnauthorized) + json.NewEncoder(w).Encode(ChatResponse{Error: "unauthorized"}) + return + } + + if !s.checkRateLimit(r) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusTooManyRequests) + json.NewEncoder(w).Encode(ChatResponse{Error: "rate limit exceeded"}) + return + } + + if r.Method == http.MethodPost { + s.handlePostChat(w, r) + return + } else if r.Method == http.MethodGet { + s.handleGetChat(w, r) + return + } + + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusMethodNotAllowed) + json.NewEncoder(w).Encode(ChatResponse{Error: "method not allowed, use POST or GET"}) +} + +func (s *Server) handlePostChat(w http.ResponseWriter, r *http.Request) { + s.mu.RLock() + chatFunc := s.chatFunc + s.mu.RUnlock() + + if chatFunc == nil { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusServiceUnavailable) + json.NewEncoder(w).Encode(ChatResponse{Error: "chat not configured"}) + return + } + + var req ChatRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusBadRequest) + json.NewEncoder(w).Encode(ChatResponse{Error: "invalid JSON: " + err.Error()}) + return + } + if req.Message == "" { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusBadRequest) + json.NewEncoder(w).Encode(ChatResponse{Error: "message field is required"}) + return + } + + sessionID := req.SessionID + if sessionID == "" && req.ChatID != "" { + sessionID = req.ChatID + } + + chatID := req.ChatID + if chatID == "" { + // Try to extract ChatID/TenantID from common headers + // These are ordered by specificity/reliability + headers := []string{ + "X-PicoClaw-Chat-ID", + "X-MS-CONVERSATION-ID", // Teams Conversation ID + "X-MS-TENANT-ID", // Teams Tenant ID + "X-User-ID", + "X-Session-ID", + "X-MS-CLIENT-PRINCIPAL-ID", // Azure App Service / Container Apps (EasyAuth) + "X-MS-CLIENT-PRINCIPAL-NAME", // Azure App Service Email/Username + "Ocp-Apim-Subscription-Id", // Azure APIM (if configured) + } + + for _, h := range headers { + if val := r.Header.Get(h); val != "" { + chatID = val + break + } + } + + // Fallback to SessionID if provided in body, otherwise empty (global) + if chatID == "" { + chatID = req.SessionID + } + } + chatID = s.sanitizeID(chatID) + sessionID = s.sanitizeID(sessionID) + + if chatID != "" { + logger.InfoCF("api", "Resolved isolation ID for request", map[string]any{ + "chat_id": chatID, + "session_id": sessionID, + }) + } else { + // Log all headers for debugging (excluding sensitive ones) + headers := make(map[string]string) + for k, v := range r.Header { + if k == "Authorization" || k == "X-Api-Key" || k == "Ocp-Apim-Subscription-Key" { + headers[k] = "REDACTED" + } else if len(v) > 0 { + headers[k] = v[0] + } + } + logger.DebugCF("api", "Chat request received without explicit ChatID. Checking headers...", map[string]any{ + "headers": headers, + }) + } + + if sessionID == "" { + sessionID = fmt.Sprintf("chat-%d", time.Now().UnixNano()) + } else { + // Even if provided, sanitize the user-provided sessionID again to be sure + sessionID = s.sanitizeID(sessionID) + } + + // Initialize status + s.chatResultsMu.Lock() + s.chatResults[sessionID] = &chatStatus{ + CreatedAt: time.Now(), + } + s.chatResultsMu.Unlock() + + // Start processing in background + go func() { + // Use a long-running context for the chat call, but don't bind to r.Context() + // which will be canceled when this request finishes. + ctx := context.Background() + logger.Debugf("Starting async chat for session %s", sessionID) + reply, err := chatFunc(ctx, req.Message, sessionID, chatID) + + s.chatResultsMu.Lock() + defer s.chatResultsMu.Unlock() + if result, ok := s.chatResults[sessionID]; ok { + result.Response = reply + result.Error = err + result.Done = true + logger.Debugf("Finished async chat for session %s (err=%v)", sessionID, err) + } + }() + + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusAccepted) + json.NewEncoder(w).Encode(ChatResponse{ + SessionID: sessionID, + Status: "pending", + }) +} + +func (s *Server) handleGetChat(w http.ResponseWriter, r *http.Request) { + sessionID := r.URL.Query().Get("session_id") + if sessionID == "" { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusBadRequest) + json.NewEncoder(w).Encode(ChatResponse{Error: "session_id query parameter is required"}) + return + } + + s.chatResultsMu.RLock() + result, ok := s.chatResults[sessionID] + if !ok { + s.chatResultsMu.RUnlock() + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusNotFound) + json.NewEncoder(w).Encode(ChatResponse{Error: "session not found"}) + return + } + + // Read fields while holding the lock to avoid race conditions + done := result.Done + response := result.Response + errVal := result.Error + s.chatResultsMu.RUnlock() + + if !done { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + json.NewEncoder(w).Encode(ChatResponse{ + SessionID: sessionID, + Status: "pending", + }) + return + } + + if errVal != nil { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusInternalServerError) + json.NewEncoder(w).Encode(ChatResponse{ + SessionID: sessionID, + Status: "error", + Error: errVal.Error(), + }) + return + } + + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + json.NewEncoder(w).Encode(ChatResponse{ + SessionID: sessionID, + Status: "completed", + Response: response, + }) +} + +func (s *Server) taskCleanupLoop() { + ticker := time.NewTicker(10 * time.Minute) + defer ticker.Stop() + + for range ticker.C { + s.chatResultsMu.Lock() + now := time.Now() + for id, status := range s.chatResults { + // Keep pending tasks for 2 hours, completed/error for 1 hour + expiry := time.Hour + if !status.Done { + expiry = 2 * time.Hour + } + + if now.Sub(status.CreatedAt) > expiry { + delete(s.chatResults, id) + logger.Debugf("Cleaned up expired chat session %s", id) + } + } + s.chatResultsMu.Unlock() + } +} + +func (s *Server) sanitizeID(id string) string { + if len(id) > 128 { + id = id[:128] + } + + result := make([]rune, 0, len(id)) + for _, r := range id { + if (r >= 'a' && r <= 'z') || (r >= 'A' && r <= 'Z') || (r >= '0' && r <= '9') || r == '_' || r == '-' { + result = append(result, r) + } else { + result = append(result, '_') + } + } + return string(result) +} + +func (s *Server) checkRateLimit(r *http.Request) bool { + // Simple rate limit: 1 request per second per ID or IP + // This is defensive against automated spamming. + key := r.Header.Get("X-PicoClaw-Chat-ID") + if key == "" { + key = r.RemoteAddr + // Strip port if present + if i := strings.LastIndex(key, ":"); i != -1 { + key = key[:i] + } + } + + if val, ok := s.rateLimits.Load(key); ok { + lastAccess := val.(time.Time) + if time.Since(lastAccess) < time.Second { + return false + } + } + + s.rateLimits.Store(key, time.Now()) + return true } func statusString(ok bool) string { @@ -235,16 +590,3 @@ func statusString(ok bool) string { } return "fail" } - -// extractBearerToken returns the token from an "Authorization: Bearer " header, -// or the empty string if the header is missing or malformed. -func extractBearerToken(header string) string { - const prefix = "Bearer " - if len(header) < len(prefix) { - return "" - } - if header[:len(prefix)] != prefix { - return "" - } - return header[len(prefix):] -}