diff --git a/cmd/picoclaw/internal/channel/start_test.go b/cmd/picoclaw/internal/channel/start_test.go index bbde329b6..0ba9cef2d 100644 --- a/cmd/picoclaw/internal/channel/start_test.go +++ b/cmd/picoclaw/internal/channel/start_test.go @@ -5,6 +5,8 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + + "github.com/sipeed/picoclaw/pkg/utils" ) func TestNewStartCommand(t *testing.T) { @@ -40,5 +42,6 @@ func TestStartCommandPreRunE_NoTruncateWithDebug(t *testing.T) { require.NoError(t, err) err = cmd.PreRunE(cmd, nil) + defer utils.SetDisableTruncation(false) assert.NoError(t, err) } diff --git a/docs/configuration.md b/docs/configuration.md index 403441f0c..9f39397b2 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -41,7 +41,7 @@ Use `picoclaw channel start` when you want to run enabled channels and AgentLoop # Start all enabled channels from config.json picoclaw channel start -# Same startup checks as gateway command +# Shares default-model/provider validation and --allow-empty behavior with gateway picoclaw channel start --allow-empty picoclaw channel start --debug ``` @@ -49,9 +49,11 @@ picoclaw channel start --debug Compared with `picoclaw gateway`, channel-only runtime: - Keeps MessageBus + AgentLoop + ChannelManager (full chat processing path) -- Starts shared HTTP server for channel webhooks and `/health`/`/ready` +- Attempts to start shared HTTP server for channel webhooks and `/health`/`/ready` (best-effort; for loopback hosts, fallback remains loopback-only and may be disabled if binding fails) - Does **not** start Cron service, Heartbeat service, Device service -- Does **not** enable config hot reload or `/reload` +- Does **not** enable config hot reload; `/reload` remains exposed by the shared health server but returns 503 (`reload not configured`) in channel-only mode + +Note: Because shared HTTP startup is best-effort, webhook-based channels may be unavailable when no usable host/port can be bound. Validation note: channel-only runtime is channel-agnostic by design. Current runtime validation is primarily focused on QQ, while other channels are continuously validated. diff --git a/docs/zh/configuration.md b/docs/zh/configuration.md index 6a42a843f..d8678206f 100644 --- a/docs/zh/configuration.md +++ b/docs/zh/configuration.md @@ -39,7 +39,7 @@ PICOCLAW_HOME=/srv/picoclaw PICOCLAW_CONFIG=/srv/picoclaw/main.json picoclaw gat # 启动 config.json 中所有已启用 channel picoclaw channel start -# 与 gateway 启动校验保持一致 +# 与 gateway 启动时的默认模型/Provider 校验及 --allow-empty 行为保持一致 picoclaw channel start --allow-empty picoclaw channel start --debug ``` @@ -47,9 +47,9 @@ picoclaw channel start --debug 与 `picoclaw gateway` 相比，channel 独立运行模式： - 保留 MessageBus + AgentLoop + ChannelManager（完整消息处理链路） -- 启动共享 HTTP 服务，用于 channel webhook 与 `/health`/`/ready` +- 尝试启动共享 HTTP 服务，用于 channel webhook 与 `/health`/`/ready`（best-effort）：若配置的 host/port 无法绑定，则会禁用共享 HTTP（当配置为 loopback 时，回退也仅限 loopback，例如 `::1` 或 `127.0.0.1`），此时 webhook 类 channel 以及 `/health`、`/ready` 将不可用 - **不会** 启动 Cron、Heartbeat、Device 服务 -- **不会** 启用配置热更新和 `/reload` +- **不会** 启用配置热更新；channel-only 模式下 `/reload` 仍可访问，但会返回 503（`reload not configured`） 验证说明：channel 独立运行模式在设计上对各渠道通用。当前运行时验证主要覆盖 QQ，其他渠道仍在持续验证中。 diff --git a/pkg/channels/manager.go b/pkg/channels/manager.go index 5fbf35ebf..41b62b769 100644 --- a/pkg/channels/manager.go +++ b/pkg/channels/manager.go @@ -77,20 +77,21 @@ type channelWorker struct { } type Manager struct { - channels map[string]Channel - workers map[string]*channelWorker - bus *bus.MessageBus - config *config.Config - mediaStore media.MediaStore - dispatchTask *asyncTask - mux *dynamicServeMux - httpServer *http.Server - mu sync.RWMutex - placeholders sync.Map // "channel:chatID" → placeholderID (string) - typingStops sync.Map // "channel:chatID" → func() - reactionUndos sync.Map // "channel:chatID" → reactionEntry - streamActive sync.Map // "channel:chatID" → true (set when streamer.Finalize sent the message) - channelHashes map[string]string // channel name → config hash + channels map[string]Channel + workers map[string]*channelWorker + bus *bus.MessageBus + config *config.Config + mediaStore media.MediaStore + dispatchTask *asyncTask + mux *dynamicServeMux + httpServer *http.Server + httpServerErrors chan error + mu sync.RWMutex + placeholders sync.Map // "channel:chatID" → placeholderID (string) + typingStops sync.Map // "channel:chatID" → func() + reactionUndos sync.Map // "channel:chatID" → reactionEntry + streamActive sync.Map // "channel:chatID" → true (set when streamer.Finalize sent the message) + channelHashes map[string]string // channel name → config hash } type asyncTask struct { @@ -452,6 +453,7 @@ func (m *Manager) SetupHTTPServer(addr string, healthServer *health.Server) { ReadTimeout: 30 * time.Second, WriteTimeout: 30 * time.Second, } + m.httpServerErrors = make(chan error, 1) } // registerHTTPHandlersLocked registers webhook and health-check handlers for @@ -541,12 +543,18 @@ func (m *Manager) StartAll(ctx context.Context) error { // Start shared HTTP server if configured if m.httpServer != nil { + httpServer := m.httpServer + httpServerErrors := m.httpServerErrors go func() { logger.InfoCF("channels", "Shared HTTP server listening", map[string]any{ - "addr": m.httpServer.Addr, + "addr": httpServer.Addr, }) - if err := m.httpServer.ListenAndServe(); err != nil && err != http.ErrServerClosed { - logger.FatalCF("channels", "Shared HTTP server error", map[string]any{ + if err := httpServer.ListenAndServe(); err != nil && err != http.ErrServerClosed { + select { + case httpServerErrors <- err: + default: + } + logger.ErrorCF("channels", "Shared HTTP server error", map[string]any{ "error": err.Error(), }) } @@ -557,6 +565,12 @@ func (m *Manager) StartAll(ctx context.Context) error { return nil } +func (m *Manager) HTTPServerErrors() <-chan error { + m.mu.RLock() + defer m.mu.RUnlock() + return m.httpServerErrors +} + func (m *Manager) StopAll(ctx context.Context) error { m.mu.Lock() defer m.mu.Unlock() @@ -573,6 +587,7 @@ func (m *Manager) StopAll(ctx context.Context) error { }) } m.httpServer = nil + m.httpServerErrors = nil } // Cancel dispatcher diff --git a/pkg/gateway/channel_only.go b/pkg/gateway/channel_only.go index 1bd92b303..fef4a067c 100644 --- a/pkg/gateway/channel_only.go +++ b/pkg/gateway/channel_only.go @@ -29,11 +29,13 @@ const ( ) type channelServices struct { - MediaStore media.MediaStore - ChannelManager *channels.Manager - HealthServer *health.Server - ListenHost string - ListenPort int + MediaStore media.MediaStore + ChannelManager *channels.Manager + HealthServer *health.Server + VoiceAgentCancel context.CancelFunc + ListenHost string + ListenPort int + ListenAddr string } // RunChannelsOnly starts channel and agent loop runtime without gateway side services. @@ -46,7 +48,7 @@ func RunChannelsOnly(debug bool, homePath, configPath string, allowEmptyStartup defer panicFunc() if err = logger.EnableFileLogging(filepath.Join(homePath, logPath, channelLogFile)); err != nil { - panic(fmt.Sprintf("error enabling file logging: %v", err)) + return fmt.Errorf("error enabling file logging: %w", err) } defer logger.DisableFileLogging() @@ -55,21 +57,24 @@ func RunChannelsOnly(debug bool, homePath, configPath string, allowEmptyStartup return fmt.Errorf("error loading config: %w", err) } - logger.SetLevelFromString(cfg.Gateway.LogLevel) + if err = preCheckConfig(cfg); err != nil { + return fmt.Errorf("invalid config: %w", err) + } if debug { logger.SetLevel(logger.DEBUG) fmt.Println("🔍 Debug mode enabled") + } else { + effectiveLogLevel := config.EffectiveGatewayLogLevel(cfg) + logger.SetLevelFromString(effectiveLogLevel) + logger.Infof("Log level set to %q", effectiveLogLevel) } - provider, modelID, err := createStartupProvider(cfg, allowEmptyStartup) + provider, modelID, err := createStartupProviderForRuntime(cfg, allowEmptyStartup, "channel-only runtime") if err != nil { return fmt.Errorf("error creating provider: %w", err) } - if allowEmptyStartup { - fmt.Println(" ⚠ Channel-only runtime started in limited mode") - } if modelID != "" { cfg.Agents.Defaults.ModelName = modelID } @@ -79,16 +84,36 @@ func RunChannelsOnly(debug bool, homePath, configPath string, allowEmptyStartup fmt.Println("\n📦 Agent Status:") startupInfo := agentLoop.GetStartupInfo() - toolsInfo := startupInfo["tools"].(map[string]any) - skillsInfo := startupInfo["skills"].(map[string]any) - fmt.Printf(" • Tools: %d loaded\n", toolsInfo["count"]) - fmt.Printf(" • Skills: %d/%d available\n", skillsInfo["available"], skillsInfo["total"]) + + var toolsCount int + if toolsRaw, ok := startupInfo["tools"].(map[string]any); ok && toolsRaw != nil { + if v, ok := toolsRaw["count"].(int); ok { + toolsCount = v + } + } + + var skillsAvailable, skillsTotal int + if skillsRaw, ok := startupInfo["skills"].(map[string]any); ok && skillsRaw != nil { + if v, ok := skillsRaw["available"].(int); ok { + skillsAvailable = v + } + if v, ok := skillsRaw["total"].(int); ok { + skillsTotal = v + } + } + + if toolsCount == 0 && skillsAvailable == 0 && skillsTotal == 0 { + fmt.Println(" • Agent startup info not available") + } else { + fmt.Printf(" • Tools: %d loaded\n", toolsCount) + fmt.Printf(" • Skills: %d/%d available\n", skillsAvailable, skillsTotal) + } logger.InfoCF("agent", "Agent initialized", map[string]any{ - "tools_count": toolsInfo["count"], - "skills_total": skillsInfo["total"], - "skills_available": skillsInfo["available"], + "tools_count": toolsCount, + "skills_total": skillsTotal, + "skills_available": skillsAvailable, }) runningServices, err := setupAndStartChannelServices(cfg, agentLoop, msgBus) @@ -96,8 +121,15 @@ func RunChannelsOnly(debug bool, homePath, configPath string, allowEmptyStartup return err } - if runningServices.ListenHost != "" { - fmt.Printf("✓ Channel runtime started on %s:%d\n", runningServices.ListenHost, runningServices.ListenPort) + if runningServices.HealthServer != nil { + if runningServices.ListenHost == "" { + fmt.Printf( + "✓ Channel runtime started (shared HTTP server enabled on all interfaces, port %d)\n", + runningServices.ListenPort, + ) + } else { + fmt.Printf("✓ Channel runtime started on %s\n", runningServices.ListenAddr) + } } else { fmt.Println("✓ Channel runtime started (shared HTTP server disabled)") } @@ -108,13 +140,20 @@ func RunChannelsOnly(debug bool, homePath, configPath string, allowEmptyStartup go agentLoop.Run(ctx) + httpErrCh := runningServices.ChannelManager.HTTPServerErrors() sigChan := make(chan os.Signal, 1) signal.Notify(sigChan, os.Interrupt, syscall.SIGTERM) - <-sigChan - logger.Info("Shutting down channel runtime...") - shutdownChannelRuntime(runningServices, agentLoop, provider) - return nil + select { + case <-sigChan: + logger.Info("Shutting down channel runtime...") + shutdownChannelRuntime(runningServices, agentLoop, provider) + return nil + case err := <-httpErrCh: + logger.Errorf("Shared HTTP server stopped in channel-only mode: %v", err) + shutdownChannelRuntime(runningServices, agentLoop, provider) + return fmt.Errorf("shared HTTP server failed: %w", err) + } } func setupAndStartChannelServices( @@ -145,7 +184,9 @@ func setupAndStartChannelServices( agentLoop.SetChannelManager(runningServices.ChannelManager) agentLoop.SetMediaStore(runningServices.MediaStore) - if transcriber := asr.DetectTranscriber(cfg); transcriber != nil { + var transcriber asr.Transcriber + transcriber = asr.DetectTranscriber(cfg) + if transcriber != nil { agentLoop.SetTranscriber(transcriber) logger.InfoCF("voice", "Transcription enabled (agent-level)", map[string]any{"provider": transcriber.Name()}) } @@ -168,6 +209,7 @@ func setupAndStartChannelServices( addr := net.JoinHostPort(listenHost, strconv.Itoa(cfg.Gateway.Port)) runningServices.ListenHost = listenHost runningServices.ListenPort = cfg.Gateway.Port + runningServices.ListenAddr = addr runningServices.HealthServer = health.NewServer(listenHost, cfg.Gateway.Port, "") runningServices.ChannelManager.SetupHTTPServer(addr, runningServices.HealthServer) } @@ -179,12 +221,29 @@ func setupAndStartChannelServices( return nil, fmt.Errorf("error starting channels: %w", err) } - if runningServices.ListenHost != "" { - fmt.Printf( - "✓ Health endpoints available at http://%s:%d/health and /ready\n", - runningServices.ListenHost, - runningServices.ListenPort, - ) + if transcriber != nil { + vaCtx, vaCancel := context.WithCancel(context.Background()) + runningServices.VoiceAgentCancel = vaCancel + voiceAgent := asr.NewAgent(msgBus, transcriber) + voiceAgent.Start(vaCtx) + } + + if runningServices.HealthServer != nil { + runningServices.HealthServer.SetReady(true) + } + + if runningServices.HealthServer != nil { + if runningServices.ListenHost == "" { + fmt.Printf( + "✓ Health endpoints available on all interfaces at port %d (/health and /ready; /reload returns 503 when not configured)\n", + runningServices.ListenPort, + ) + } else { + fmt.Printf( + "✓ Health endpoints available at http://%s/health and /ready (/reload returns 503 when not configured)\n", + runningServices.ListenAddr, + ) + } } else { fmt.Println("⚠ Shared HTTP server disabled; /health and webhook endpoints are unavailable") } @@ -196,24 +255,40 @@ func resolveChannelOnlyListenHost(host string, port int) (string, error) { if err := probeTCPBind(host, port); err == nil { return host, nil } else if isLoopbackHost(host) { - fallbackErr := probeTCPBind("0.0.0.0", port) - if fallbackErr == nil { + // Keep loopback scope when fallback is needed to avoid widening exposure. + ipv6FallbackErr := probeTCPBind("::1", port) + if ipv6FallbackErr == nil { logger.WarnCF( "channels", - "Loopback host unavailable in channel-only mode, fallback to wildcard", + "Loopback host unavailable in channel-only mode, fallback to IPv6 loopback", map[string]any{ "host": host, "port": port, }, ) - return "0.0.0.0", nil + return "::1", nil + } + + ipv4FallbackErr := probeTCPBind("127.0.0.1", port) + if ipv4FallbackErr == nil { + logger.WarnCF( + "channels", + "Loopback host unavailable in channel-only mode, fallback to IPv4 loopback", + map[string]any{ + "host": host, + "port": port, + }, + ) + return "127.0.0.1", nil } return "", fmt.Errorf( - "bind fallback 0.0.0.0:%d failed after %s:%d failed: %w (original error: %v)", + "bind fallback [::1]:%d and 127.0.0.1:%d failed after %s:%d failed: ipv6 error: %v; ipv4 error: %v; original error: %v", + port, port, host, port, - fallbackErr, + ipv6FallbackErr, + ipv4FallbackErr, err, ) } else { @@ -250,6 +325,14 @@ func shutdownChannelRuntime( } if runningServices != nil { + if runningServices.HealthServer != nil { + runningServices.HealthServer.SetReady(false) + } + + if runningServices.VoiceAgentCancel != nil { + runningServices.VoiceAgentCancel() + } + shutdownCtx, cancel := context.WithTimeout(context.Background(), gracefulShutdownTimeout) defer cancel() diff --git a/pkg/gateway/channel_only_test.go b/pkg/gateway/channel_only_test.go new file mode 100644 index 000000000..3dffe1c8c --- /dev/null +++ b/pkg/gateway/channel_only_test.go @@ -0,0 +1,13 @@ +package gateway + +import "testing" + +func TestResolveChannelOnlyListenHostRejectsNonLoopbackFallback(t *testing.T) { + host, err := resolveChannelOnlyListenHost("256.256.256.256", 1) + if err == nil { + t.Fatal("expected error for invalid non-loopback host") + } + if host != "" { + t.Fatalf("host = %q, want empty", host) + } +} diff --git a/pkg/gateway/gateway.go b/pkg/gateway/gateway.go index 8065a0795..37716d7bb 100644 --- a/pkg/gateway/gateway.go +++ b/pkg/gateway/gateway.go @@ -288,12 +288,26 @@ func executeReload( func createStartupProvider( cfg *config.Config, allowEmptyStartup bool, +) (providers.LLMProvider, string, error) { + return createStartupProviderForRuntime(cfg, allowEmptyStartup, "gateway") +} + +func createStartupProviderForRuntime( + cfg *config.Config, + allowEmptyStartup bool, + runtimeName string, ) (providers.LLMProvider, string, error) { modelName := cfg.Agents.Defaults.GetModelName() if modelName == "" && allowEmptyStartup { - reason := "no default model configured; gateway started in limited mode" + runtimeName = strings.TrimSpace(runtimeName) + if runtimeName == "" { + runtimeName = "gateway" + } + + reason := fmt.Sprintf("no default model configured; %s started in limited mode", runtimeName) fmt.Printf("⚠ Warning: %s\n", reason) - logger.WarnCF("gateway", "Gateway started without default model", map[string]any{ + logger.WarnCF("runtime", "Runtime started without default model", map[string]any{ + "runtime": runtimeName, "limited_mode": true, }) return &startupBlockedProvider{reason: reason}, "", nil diff --git a/pkg/health/server.go b/pkg/health/server.go index 2602cb965..79c2916ba 100644 --- a/pkg/health/server.go +++ b/pkg/health/server.go @@ -222,7 +222,7 @@ type HandlerMux interface { } // RegisterOnMux registers /health, /ready and /reload handlers onto the given mux. -// This allows the health endpoints to be served by a shared HTTP server. +// The /reload handler checks reload availability at request time. func (s *Server) RegisterOnMux(mux HandlerMux) { mux.HandleFunc("/health", s.healthHandler) mux.HandleFunc("/ready", s.readyHandler) diff --git a/pkg/health/server_test.go b/pkg/health/server_test.go index c4982fff9..238743bbf 100644 --- a/pkg/health/server_test.go +++ b/pkg/health/server_test.go @@ -20,6 +20,12 @@ func newTestServer() *Server { return s } +func newReloadEnabledTestServer() *Server { + s := newTestServer() + s.SetReloadFunc(func() error { return nil }) + return s +} + func TestHealthHandler_ReturnsOK(t *testing.T) { s := newTestServer() req := httptest.NewRequest(http.MethodGet, "/health", nil) @@ -150,7 +156,7 @@ func TestReadyHandler_PassingCheck(t *testing.T) { } func TestReloadHandler_MethodNotAllowed(t *testing.T) { - s := newTestServer() + s := newReloadEnabledTestServer() req := httptest.NewRequest(http.MethodGet, "/reload", nil) w := httptest.NewRecorder() @@ -177,7 +183,7 @@ func TestReloadHandler_NoReloadFunc(t *testing.T) { } func TestReloadHandler_Success(t *testing.T) { - s := newTestServer() + s := newReloadEnabledTestServer() called := false s.SetReloadFunc(func() error { called = true @@ -199,7 +205,7 @@ func TestReloadHandler_Success(t *testing.T) { } func TestReloadHandler_Error(t *testing.T) { - s := newTestServer() + s := newReloadEnabledTestServer() s.SetReloadFunc(func() error { return errors.New("config parse error") }) @@ -290,6 +296,31 @@ func TestRegisterOnMux(t *testing.T) { if w.Code != http.StatusOK { t.Errorf("/ready on custom mux = %d, want %d", w.Code, http.StatusOK) } + + req = httptest.NewRequest(http.MethodPost, "/reload", nil) + req.Header.Set("Authorization", "Bearer test") + w = httptest.NewRecorder() + mux.ServeHTTP(w, req) + if w.Code != http.StatusServiceUnavailable { + t.Errorf("/reload on mux without reload = %d, want %d", w.Code, http.StatusServiceUnavailable) + } +} + +func TestRegisterOnMux_WithReload(t *testing.T) { + s := newReloadEnabledTestServer() + s.SetReady(true) + + mux := http.NewServeMux() + s.RegisterOnMux(mux) + + req := httptest.NewRequest(http.MethodPost, "/reload", nil) + req.Header.Set("Authorization", "Bearer test") + w := httptest.NewRecorder() + mux.ServeHTTP(w, req) + + if w.Code != http.StatusOK { + t.Errorf("/reload on mux with reload = %d, want %d", w.Code, http.StatusOK) + } } func TestNewServer(t *testing.T) {