feat(channel): add channel-only runtime command
This commit is contained in:
parent
415abc8cd4
commit
0340805601
9 changed files with 438 additions and 0 deletions
19
cmd/picoclaw/internal/channel/command.go
Normal file
19
cmd/picoclaw/internal/channel/command.go
Normal file
|
|
@ -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
|
||||
}
|
||||
29
cmd/picoclaw/internal/channel/command_test.go
Normal file
29
cmd/picoclaw/internal/channel/command_test.go
Normal file
|
|
@ -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())
|
||||
}
|
||||
51
cmd/picoclaw/internal/channel/start.go
Normal file
51
cmd/picoclaw/internal/channel/start.go
Normal file
|
|
@ -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
|
||||
}
|
||||
44
cmd/picoclaw/internal/channel/start_test.go
Normal file
44
cmd/picoclaw/internal/channel/start_test.go
Normal file
|
|
@ -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)
|
||||
}
|
||||
|
|
@ -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(),
|
||||
|
|
|
|||
|
|
@ -36,6 +36,7 @@ func TestNewPicoclawCommand(t *testing.T) {
|
|||
allowedCommands := []string{
|
||||
"agent",
|
||||
"auth",
|
||||
"channel",
|
||||
"cron",
|
||||
"gateway",
|
||||
"migrate",
|
||||
|
|
|
|||
|
|
@ -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`.
|
||||
|
|
|
|||
|
|
@ -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` 中配置:
|
||||
|
|
|
|||
248
pkg/gateway/channel_only.go
Normal file
248
pkg/gateway/channel_only.go
Normal file
|
|
@ -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()
|
||||
}
|
||||
}
|
||||
Loading…
Add table
Reference in a new issue