diff --git a/cmd/picoclaw/internal/channel/command.go b/cmd/picoclaw/internal/channel/command.go new file mode 100644 index 000000000..717b997f7 --- /dev/null +++ b/cmd/picoclaw/internal/channel/command.go @@ -0,0 +1,19 @@ +package channel + +import "github.com/spf13/cobra" + +func NewChannelCommand() *cobra.Command { + cmd := &cobra.Command{ + Use: "channel", + Aliases: []string{"ch"}, + Short: "Manage channel runtimes", + Args: cobra.NoArgs, + RunE: func(cmd *cobra.Command, _ []string) error { + return cmd.Help() + }, + } + + cmd.AddCommand(newStartCommand()) + + return cmd +} diff --git a/cmd/picoclaw/internal/channel/command_test.go b/cmd/picoclaw/internal/channel/command_test.go new file mode 100644 index 000000000..22726f972 --- /dev/null +++ b/cmd/picoclaw/internal/channel/command_test.go @@ -0,0 +1,29 @@ +package channel + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestNewChannelCommand(t *testing.T) { + cmd := NewChannelCommand() + + require.NotNil(t, cmd) + + assert.Equal(t, "channel", cmd.Use) + assert.Equal(t, "Manage channel runtimes", cmd.Short) + + assert.Len(t, cmd.Aliases, 1) + assert.True(t, cmd.HasAlias("ch")) + + assert.Nil(t, cmd.Run) + assert.NotNil(t, cmd.RunE) + + assert.True(t, cmd.HasSubCommands()) + startCmd, _, err := cmd.Find([]string{"start"}) + require.NoError(t, err) + require.NotNil(t, startCmd) + assert.Equal(t, "start", startCmd.Name()) +} diff --git a/cmd/picoclaw/internal/channel/start.go b/cmd/picoclaw/internal/channel/start.go new file mode 100644 index 000000000..c8ddb5420 --- /dev/null +++ b/cmd/picoclaw/internal/channel/start.go @@ -0,0 +1,51 @@ +package channel + +import ( + "fmt" + + "github.com/spf13/cobra" + + "github.com/sipeed/picoclaw/cmd/picoclaw/internal" + "github.com/sipeed/picoclaw/pkg/gateway" + "github.com/sipeed/picoclaw/pkg/logger" + "github.com/sipeed/picoclaw/pkg/utils" +) + +func newStartCommand() *cobra.Command { + var debug bool + var noTruncate bool + var allowEmpty bool + + cmd := &cobra.Command{ + Use: "start", + Short: "Start channels without gateway side services", + Args: cobra.NoArgs, + PreRunE: func(_ *cobra.Command, _ []string) error { + if noTruncate && !debug { + return fmt.Errorf("the --no-truncate option can only be used in conjunction with --debug (-d)") + } + + if noTruncate { + utils.SetDisableTruncation(true) + logger.Info("String truncation is globally disabled via 'no-truncate' flag") + } + + return nil + }, + RunE: func(_ *cobra.Command, _ []string) error { + return gateway.RunChannelsOnly(debug, internal.GetPicoclawHome(), internal.GetConfigPath(), allowEmpty) + }, + } + + cmd.Flags().BoolVarP(&debug, "debug", "d", false, "Enable debug logging") + cmd.Flags().BoolVarP(&noTruncate, "no-truncate", "T", false, "Disable string truncation in debug logs") + cmd.Flags().BoolVarP( + &allowEmpty, + "allow-empty", + "E", + false, + "Continue starting even when no default model is configured", + ) + + return cmd +} diff --git a/cmd/picoclaw/internal/channel/start_test.go b/cmd/picoclaw/internal/channel/start_test.go new file mode 100644 index 000000000..bbde329b6 --- /dev/null +++ b/cmd/picoclaw/internal/channel/start_test.go @@ -0,0 +1,44 @@ +package channel + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestNewStartCommand(t *testing.T) { + cmd := newStartCommand() + + require.NotNil(t, cmd) + + assert.Equal(t, "start", cmd.Use) + assert.Equal(t, "Start channels without gateway side services", cmd.Short) + assert.NotNil(t, cmd.RunE) + assert.NotNil(t, cmd.PreRunE) + + assert.NotNil(t, cmd.Flags().Lookup("debug")) + assert.NotNil(t, cmd.Flags().Lookup("allow-empty")) + assert.NotNil(t, cmd.Flags().Lookup("no-truncate")) +} + +func TestStartCommandPreRunE_NoTruncateRequiresDebug(t *testing.T) { + cmd := newStartCommand() + + err := cmd.ParseFlags([]string{"--no-truncate"}) + require.NoError(t, err) + + err = cmd.PreRunE(cmd, nil) + assert.Error(t, err) + assert.Contains(t, err.Error(), "--no-truncate") +} + +func TestStartCommandPreRunE_NoTruncateWithDebug(t *testing.T) { + cmd := newStartCommand() + + err := cmd.ParseFlags([]string{"--debug", "--no-truncate"}) + require.NoError(t, err) + + err = cmd.PreRunE(cmd, nil) + assert.NoError(t, err) +} diff --git a/cmd/picoclaw/main.go b/cmd/picoclaw/main.go index 48dffbb33..36259ca66 100644 --- a/cmd/picoclaw/main.go +++ b/cmd/picoclaw/main.go @@ -15,6 +15,7 @@ import ( "github.com/sipeed/picoclaw/cmd/picoclaw/internal" "github.com/sipeed/picoclaw/cmd/picoclaw/internal/agent" "github.com/sipeed/picoclaw/cmd/picoclaw/internal/auth" + "github.com/sipeed/picoclaw/cmd/picoclaw/internal/channel" "github.com/sipeed/picoclaw/cmd/picoclaw/internal/cron" "github.com/sipeed/picoclaw/cmd/picoclaw/internal/gateway" "github.com/sipeed/picoclaw/cmd/picoclaw/internal/migrate" @@ -40,6 +41,7 @@ func NewPicoclawCommand() *cobra.Command { onboard.NewOnboardCommand(), agent.NewAgentCommand(), auth.NewAuthCommand(), + channel.NewChannelCommand(), gateway.NewGatewayCommand(), status.NewStatusCommand(), cron.NewCronCommand(), diff --git a/cmd/picoclaw/main_test.go b/cmd/picoclaw/main_test.go index cb221dece..19f8264ba 100644 --- a/cmd/picoclaw/main_test.go +++ b/cmd/picoclaw/main_test.go @@ -36,6 +36,7 @@ func TestNewPicoclawCommand(t *testing.T) { allowedCommands := []string{ "agent", "auth", + "channel", "cron", "gateway", "migrate", diff --git a/docs/configuration.md b/docs/configuration.md index 58930cbfa..ba3e29adc 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -33,6 +33,28 @@ PICOCLAW_HOME=/opt/picoclaw picoclaw agent PICOCLAW_HOME=/srv/picoclaw PICOCLAW_CONFIG=/srv/picoclaw/main.json picoclaw gateway ``` +### Channel-Only Runtime + +Use `picoclaw channel start` when you want to run enabled channels and AgentLoop without starting Gateway side services. + +```bash +# Start all enabled channels from config.json +picoclaw channel start + +# Same startup checks as gateway command +picoclaw channel start --allow-empty +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` +- Does **not** start Cron service, Heartbeat service, Device service +- Does **not** enable config hot reload or `/reload` + +Use `picoclaw gateway` if you need the full control plane and background services. + ### Gateway Log Level `gateway.log_level` controls Gateway log verbosity and is configurable in `config.json`. diff --git a/docs/zh/configuration.md b/docs/zh/configuration.md index a405df09c..392ae5f74 100644 --- a/docs/zh/configuration.md +++ b/docs/zh/configuration.md @@ -31,6 +31,28 @@ PICOCLAW_HOME=/opt/picoclaw picoclaw agent PICOCLAW_HOME=/srv/picoclaw PICOCLAW_CONFIG=/srv/picoclaw/main.json picoclaw gateway ``` +### Channel 独立运行模式 + +当你希望在不启动 gateway 侧服务的情况下,仅运行已启用的 channel 与 AgentLoop 时,可使用 `picoclaw channel start`。 + +```bash +# 启动 config.json 中所有已启用 channel +picoclaw channel start + +# 与 gateway 启动校验保持一致 +picoclaw channel start --allow-empty +picoclaw channel start --debug +``` + +与 `picoclaw gateway` 相比,channel 独立运行模式: + +- 保留 MessageBus + AgentLoop + ChannelManager(完整消息处理链路) +- 启动共享 HTTP 服务,用于 channel webhook 与 `/health`/`/ready` +- **不会** 启动 Cron、Heartbeat、Device 服务 +- **不会** 启用配置热更新和 `/reload` + +若需要完整控制面与后台服务,请使用 `picoclaw gateway`。 + ### Gateway 日志等级 `gateway.log_level` 控制 Gateway 的日志详细程度,可在 `config.json` 中配置: diff --git a/pkg/gateway/channel_only.go b/pkg/gateway/channel_only.go new file mode 100644 index 000000000..28046c393 --- /dev/null +++ b/pkg/gateway/channel_only.go @@ -0,0 +1,248 @@ +package gateway + +import ( + "context" + "fmt" + "net" + "os" + "os/signal" + "path/filepath" + "strconv" + "strings" + "syscall" + "time" + + "github.com/sipeed/picoclaw/pkg/agent" + "github.com/sipeed/picoclaw/pkg/bus" + "github.com/sipeed/picoclaw/pkg/channels" + "github.com/sipeed/picoclaw/pkg/config" + "github.com/sipeed/picoclaw/pkg/health" + "github.com/sipeed/picoclaw/pkg/logger" + "github.com/sipeed/picoclaw/pkg/media" + "github.com/sipeed/picoclaw/pkg/providers" + "github.com/sipeed/picoclaw/pkg/voice" +) + +const ( + channelPanicFile = "channel_panic.log" + channelLogFile = "channel.log" +) + +type channelServices struct { + MediaStore media.MediaStore + ChannelManager *channels.Manager + HealthServer *health.Server + ListenHost string + ListenPort int +} + +// RunChannelsOnly starts channel and agent loop runtime without gateway side services. +func RunChannelsOnly(debug bool, homePath, configPath string, allowEmptyStartup bool) error { + panicPath := filepath.Join(homePath, logPath, channelPanicFile) + panicFunc, err := logger.InitPanic(panicPath) + if err != nil { + return fmt.Errorf("error initializing panic log: %w", err) + } + defer panicFunc() + + if err = logger.EnableFileLogging(filepath.Join(homePath, logPath, channelLogFile)); err != nil { + panic(fmt.Sprintf("error enabling file logging: %v", err)) + } + defer logger.DisableFileLogging() + + cfg, err := config.LoadConfig(configPath) + if err != nil { + return fmt.Errorf("error loading config: %w", err) + } + + logger.SetLevelFromString(cfg.Gateway.LogLevel) + + if debug { + logger.SetLevel(logger.DEBUG) + fmt.Println("🔍 Debug mode enabled") + } + + provider, modelID, err := createStartupProvider(cfg, allowEmptyStartup) + if err != nil { + return fmt.Errorf("error creating provider: %w", err) + } + + if modelID != "" { + cfg.Agents.Defaults.ModelName = modelID + } + + msgBus := bus.NewMessageBus() + agentLoop := agent.NewAgentLoop(cfg, msgBus, provider) + + 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"]) + + logger.InfoCF("agent", "Agent initialized", + map[string]any{ + "tools_count": toolsInfo["count"], + "skills_total": skillsInfo["total"], + "skills_available": skillsInfo["available"], + }) + + runningServices, err := setupAndStartChannelServices(cfg, agentLoop, msgBus) + if err != nil { + return err + } + + if runningServices.ListenHost != "" { + fmt.Printf("✓ Channel runtime started on %s:%d\n", runningServices.ListenHost, runningServices.ListenPort) + } else { + fmt.Println("✓ Channel runtime started (shared HTTP server disabled)") + } + fmt.Println("Press Ctrl+C to stop") + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + go agentLoop.Run(ctx) + + 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 +} + +func setupAndStartChannelServices( + cfg *config.Config, + agentLoop *agent.AgentLoop, + msgBus *bus.MessageBus, +) (*channelServices, error) { + runningServices := &channelServices{} + + runningServices.MediaStore = media.NewFileMediaStoreWithCleanup(media.MediaCleanerConfig{ + Enabled: cfg.Tools.MediaCleanup.Enabled, + MaxAge: time.Duration(cfg.Tools.MediaCleanup.MaxAge) * time.Minute, + Interval: time.Duration(cfg.Tools.MediaCleanup.Interval) * time.Minute, + }) + if fms, ok := runningServices.MediaStore.(*media.FileMediaStore); ok { + fms.Start() + } + + var err error + runningServices.ChannelManager, err = channels.NewManager(cfg, msgBus, runningServices.MediaStore) + if err != nil { + if fms, ok := runningServices.MediaStore.(*media.FileMediaStore); ok { + fms.Stop() + } + return nil, fmt.Errorf("error creating channel manager: %w", err) + } + + agentLoop.SetChannelManager(runningServices.ChannelManager) + agentLoop.SetMediaStore(runningServices.MediaStore) + + if transcriber := voice.DetectTranscriber(cfg); transcriber != nil { + agentLoop.SetTranscriber(transcriber) + logger.InfoCF("voice", "Transcription enabled (agent-level)", map[string]any{"provider": transcriber.Name()}) + } + + enabledChannels := runningServices.ChannelManager.GetEnabledChannels() + if len(enabledChannels) > 0 { + fmt.Printf("✓ Channels enabled: %s\n", enabledChannels) + } else { + fmt.Println("⚠ Warning: No channels enabled") + } + + listenHost, resolveErr := resolveChannelOnlyListenHost(cfg.Gateway.Host, cfg.Gateway.Port) + if resolveErr != nil { + logger.WarnCF("channels", "Shared HTTP server disabled in channel-only mode", map[string]any{ + "host": cfg.Gateway.Host, + "port": cfg.Gateway.Port, + "error": resolveErr.Error(), + }) + } else { + addr := net.JoinHostPort(listenHost, strconv.Itoa(cfg.Gateway.Port)) + runningServices.ListenHost = listenHost + runningServices.ListenPort = cfg.Gateway.Port + runningServices.HealthServer = health.NewServer(listenHost, cfg.Gateway.Port) + runningServices.ChannelManager.SetupHTTPServer(addr, runningServices.HealthServer) + } + + if err = runningServices.ChannelManager.StartAll(context.Background()); err != nil { + if fms, ok := runningServices.MediaStore.(*media.FileMediaStore); ok { + fms.Stop() + } + 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) + } else { + fmt.Println("⚠ Shared HTTP server disabled; /health and webhook endpoints are unavailable") + } + + return runningServices, nil +} + +func resolveChannelOnlyListenHost(host string, port int) (string, error) { + if err := probeTCPBind(host, port); err == nil { + return host, nil + } else if isLoopbackHost(host) { + if fallbackErr := probeTCPBind("0.0.0.0", port); fallbackErr == nil { + logger.WarnCF("channels", "Loopback host unavailable in channel-only mode, fallback to wildcard", map[string]any{ + "host": host, + "port": port, + }) + return "0.0.0.0", nil + } + return "", fmt.Errorf("bind %s:%d failed: %w", host, port, err) + } else { + return "", fmt.Errorf("bind %s:%d failed: %w", host, port, err) + } +} + +func probeTCPBind(host string, port int) error { + addr := net.JoinHostPort(host, strconv.Itoa(port)) + ln, err := net.Listen("tcp", addr) + if err != nil { + return err + } + _ = ln.Close() + return nil +} + +func isLoopbackHost(host string) bool { + normalized := strings.TrimSpace(strings.ToLower(host)) + if normalized == "localhost" { + return true + } + ip := net.ParseIP(normalized) + return ip != nil && ip.IsLoopback() +} + +func shutdownChannelRuntime(runningServices *channelServices, agentLoop *agent.AgentLoop, provider providers.LLMProvider) { + if cp, ok := provider.(providers.StatefulProvider); ok { + cp.Close() + } + + if runningServices != nil { + shutdownCtx, cancel := context.WithTimeout(context.Background(), gracefulShutdownTimeout) + defer cancel() + + if runningServices.ChannelManager != nil { + runningServices.ChannelManager.StopAll(shutdownCtx) + } + + if runningServices.MediaStore != nil { + if fms, ok := runningServices.MediaStore.(*media.FileMediaStore); ok { + fms.Stop() + } + } + } + + if agentLoop != nil { + agentLoop.Stop() + agentLoop.Close() + } +}