opencode/picoclaw
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

web

Browse source
This commit is contained in:
avaksru 2026-03-23 12:12:01 +03:00
parent 5477c24dbd
commit d72e8e4d91
7 changed files with 1141 additions and 1641 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

49
pkg/channels/manager.go
View file

@ -158,13 +158,8 @@ func (m *Manager) RecordReactionUndo(channel, chatID string, undo func()) {
}
// preSend handles typing stop, reaction undo, and placeholder editing before sending a message.
// Returns true if the message was edited into a placeholder (skip Send).
func (m *Manager) preSend(
ctx context.Context,
name string,
msg bus.OutboundMessage,
ch Channel,
) bool {
// Returns true if the message was already delivered (skip Send).
func (m *Manager) preSend(ctx context.Context, name string, msg bus.OutboundMessage, ch Channel) bool {
key := name + ":" + msg.ChatID
// 1. Stop typing
@ -211,11 +206,7 @@ func (m *Manager) preSend(
return false
}
func NewManager(
cfg *config.Config,
messageBus *bus.MessageBus,
store media.MediaStore,
) (*Manager, error) {
func NewManager(cfg *config.Config, messageBus *bus.MessageBus, store media.MediaStore) (*Manager, error) {
m := &Manager{
channels: make(map[string]Channel),
workers: make(map[string]*channelWorker),
@ -633,12 +624,7 @@ func (m *Manager) runWorker(ctx context.Context, name string, w *channelWorker)
// - ErrNotRunning / ErrSendFailed: permanent, no retry
// - ErrRateLimit: fixed delay retry
// - ErrTemporary / unknown: exponential backoff retry
func (m *Manager) sendWithRetry(
ctx context.Context,
name string,
w *channelWorker,
msg bus.OutboundMessage,
) {
func (m *Manager) sendWithRetry(ctx context.Context, name string, w *channelWorker, msg bus.OutboundMessage) {
// Rate limit: wait for token
if err := w.limiter.Wait(ctx); err != nil {
// ctx canceled, shutting down
@ -678,10 +664,7 @@ func (m *Manager) sendWithRetry(
}
// ErrTemporary or unknown error — exponential backoff
backoff := min(
time.Duration(float64(baseBackoff)*math.Pow(2, float64(attempt))),
maxBackoff,
)
backoff := min(time.Duration(float64(baseBackoff)*math.Pow(2, float64(attempt))), maxBackoff)
select {
case <-time.After(backoff):
case <-ctx.Done():
@ -806,21 +789,12 @@ func (m *Manager) runMediaWorker(ctx context.Context, name string, w *channelWor
// sendMediaWithRetry sends a media message through the channel with rate limiting and
// retry logic. If the channel does not implement MediaSender, it silently skips.
func (m *Manager) sendMediaWithRetry(
ctx context.Context,
name string,
w *channelWorker,
msg bus.OutboundMediaMessage,
) {
func (m *Manager) sendMediaWithRetry(ctx context.Context, name string, w *channelWorker, msg bus.OutboundMediaMessage) {
ms, ok := w.ch.(MediaSender)
if !ok {
logger.DebugCF(
"channels",
"Channel does not support MediaSender, skipping media",
map[string]any{
"channel": name,
},
)
logger.DebugCF("channels", "Channel does not support MediaSender, skipping media", map[string]any{
"channel": name,
})
return
}
@ -857,10 +831,7 @@ func (m *Manager) sendMediaWithRetry(
}
// ErrTemporary or unknown error — exponential backoff
backoff := min(
time.Duration(float64(baseBackoff)*math.Pow(2, float64(attempt))),
maxBackoff,
)
backoff := min(time.Duration(float64(baseBackoff)*math.Pow(2, float64(attempt))), maxBackoff)
select {
case <-time.After(backoff):
case <-ctx.Done():

2664
pkg/config/config.go
View file

File diff suppressed because it is too large Load diff

60
pkg/gateway/gateway.go
View file

@ -45,29 +45,20 @@ import (
)
const (
serviceShutdownTimeout = 30 * time.Second
providerReloadTimeout = 30 * time.Second
serviceShutdownTimeout = 30 * time.Second
providerReloadTimeout = 30 * time.Second
gracefulShutdownTimeout = 15 * time.Second
)
type services struct {
CronService *cron.CronService
CronService *cron.CronService
HeartbeatService *heartbeat.HeartbeatService
MediaStore media.MediaStore
ChannelManager *channels.Manager
DeviceService *devices.Service
HealthServer *health.Server
MediaStore media.MediaStore
ChannelManager *channels.Manager
DeviceService *devices.Service
HealthServer *health.Server
manualReloadChan chan struct{}
reloading atomic.Bool
reloading atomic.Bool
}
type startupBlockedProvider struct {
@ -90,11 +81,6 @@ func (p *startupBlockedProvider) GetDefaultModel() string {
// Run starts the gateway runtime using the configuration loaded from configPath.
func Run(debug bool, configPath string, allowEmptyStartup bool) error {
if debug {
logger.SetLevel(logger.DEBUG)
fmt.Println("🔍 Debug mode enabled")
}
cfg, err := config.LoadConfig(configPath)
if err != nil {
return fmt.Errorf("error loading config: %w", err)
@ -123,7 +109,6 @@ func Run(debug bool, configPath string, allowEmptyStartup bool) error {
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"])
@ -142,7 +127,6 @@ func Run(debug bool, configPath string, allowEmptyStartup bool) error {
// Setup manual reload channel for /reload endpoint
manualReloadChan := make(chan struct{}, 1)
runningServices.manualReloadChan = manualReloadChan
reloadTrigger := func() error {
if !runningServices.reloading.CompareAndSwap(false, true) {
return fmt.Errorf("reload already in progress")
@ -156,7 +140,6 @@ func Run(debug bool, configPath string, allowEmptyStartup bool) error {
return fmt.Errorf("reload already queued")
}
}
runningServices.HealthServer.SetReloadFunc(reloadTrigger)
agentLoop.SetReloadFunc(reloadTrigger)
@ -170,7 +153,6 @@ func Run(debug bool, configPath string, allowEmptyStartup bool) error {
var configReloadChan <-chan *config.Config
stopWatch := func() {}
if cfg.Gateway.HotReload {
configReloadChan, stopWatch = setupConfigWatcherPolling(configPath, debug)
logger.Info("Config hot reload enabled")
@ -186,7 +168,6 @@ func Run(debug bool, configPath string, allowEmptyStartup bool) error {
logger.Info("Shutting down...")
shutdownGateway(runningServices, agentLoop, provider, true)
return nil
case newCfg := <-configReloadChan:
if !runningServices.reloading.CompareAndSwap(false, true) {
logger.Warn("Config reload skipped: another reload is in progress")
@ -196,7 +177,6 @@ func Run(debug bool, configPath string, allowEmptyStartup bool) error {
if err != nil {
logger.Errorf("Config reload failed: %v", err)
}
case <-manualReloadChan:
logger.Info("Manual reload triggered via /reload endpoint")
newCfg, err := config.LoadConfig(configPath)
@ -246,6 +226,7 @@ func createStartupProvider(
})
return &startupBlockedProvider{reason: reason}, "", nil
}
return providers.CreateProvider(cfg)
}
@ -255,8 +236,8 @@ func setupAndStartServices(
msgBus *bus.MessageBus,
) (*services, error) {
runningServices := &services{}
execTimeout := time.Duration(cfg.Tools.Cron.ExecTimeoutMinutes) * time.Minute
execTimeout := time.Duration(cfg.Tools.Cron.ExecTimeoutMinutes) * time.Minute
var err error
runningServices.CronService, err = setupCronTool(
agentLoop,
@ -302,6 +283,7 @@ func setupAndStartServices(
}
return nil, fmt.Errorf("error creating channel manager: %w", err)
}
agentLoop.SetChannelManager(runningServices.ChannelManager)
agentLoop.SetMediaStore(runningServices.MediaStore)
@ -320,9 +302,11 @@ func setupAndStartServices(
addr := fmt.Sprintf("%s:%d", cfg.Gateway.Host, cfg.Gateway.Port)
runningServices.HealthServer = health.NewServer(cfg.Gateway.Host, cfg.Gateway.Port)
runningServices.ChannelManager.SetupHTTPServer(addr, runningServices.HealthServer)
if err = runningServices.ChannelManager.StartAll(context.Background()); err != nil {
return nil, fmt.Errorf("error starting channels: %w", err)
}
fmt.Printf(
"✓ Health endpoints available at http://%s:%d/health, /ready and /reload (POST)\n",
cfg.Gateway.Host,
@ -352,19 +336,15 @@ func stopAndCleanupServices(runningServices *services, shutdownTimeout time.Dura
if !isReload && runningServices.ChannelManager != nil {
runningServices.ChannelManager.StopAll(shutdownCtx)
}
if runningServices.DeviceService != nil {
runningServices.DeviceService.Stop()
}
if runningServices.HeartbeatService != nil {
runningServices.HeartbeatService.Stop()
}
if runningServices.CronService != nil {
runningServices.CronService.Stop()
}
if runningServices.MediaStore != nil {
if fms, ok := runningServices.MediaStore.(*media.FileMediaStore); ok {
fms.Stop()
@ -381,9 +361,12 @@ func shutdownGateway(
if cp, ok := provider.(providers.StatefulProvider); ok && fullShutdown {
cp.Close()
}
stopAndCleanupServices(runningServices, gracefulShutdownTimeout, false)
agentLoop.Stop()
agentLoop.Close()
logger.Info("✓ Gateway stopped")
}
@ -401,6 +384,7 @@ func handleConfigReload(
newModel := newCfg.Agents.Defaults.ModelName
logger.Infof(" New model is '%s', recreating provider...", newModel)
logger.Info(" Stopping all services...")
stopAndCleanupServices(runningServices, serviceShutdownTimeout, true)
@ -451,8 +435,8 @@ func restartServices(
msgBus *bus.MessageBus,
) error {
cfg := al.GetConfig()
execTimeout := time.Duration(cfg.Tools.Cron.ExecTimeoutMinutes) * time.Minute
execTimeout := time.Duration(cfg.Tools.Cron.ExecTimeoutMinutes) * time.Minute
var err error
runningServices.CronService, err = setupCronTool(
al,
@ -506,7 +490,6 @@ func restartServices(
}
addr := fmt.Sprintf("%s:%d", cfg.Gateway.Host, cfg.Gateway.Port)
// Reuse existing HealthServer to preserve reloadFunc
if runningServices.HealthServer == nil {
runningServices.HealthServer = health.NewServer(cfg.Gateway.Host, cfg.Gateway.Port)
@ -532,7 +515,6 @@ func restartServices(
transcriber := voice.DetectTranscriber(cfg)
al.SetTranscriber(transcriber)
if transcriber != nil {
logger.InfoCF("voice", "Transcription re-enabled (agent-level)", map[string]any{"provider": transcriber.Name()})
} else {
@ -553,6 +535,7 @@ func setupConfigWatcherPolling(configPath string, debug bool) (chan *config.Conf
lastModTime := getFileModTime(configPath)
lastSize := getFileSize(configPath)
ticker := time.NewTicker(2 * time.Second)
defer ticker.Stop()
@ -568,6 +551,7 @@ func setupConfigWatcherPolling(configPath string, debug bool) (chan *config.Conf
}
time.Sleep(500 * time.Millisecond)
lastModTime = currentModTime
lastSize = currentSize
@ -631,6 +615,7 @@ func setupCronTool(
cfg *config.Config,
) (*cron.CronService, error) {
cronStorePath := filepath.Join(workspace, "cron", "jobs.json")
cronService := cron.NewCronService(cronStorePath, nil)
var cronTool *tools.CronTool
@ -640,6 +625,7 @@ func setupCronTool(
if err != nil {
return nil, fmt.Errorf("critical error during CronTool initialization: %w", err)
}
agentLoop.RegisterTool(cronTool)
}
@ -663,11 +649,9 @@ func createHeartbeatHandler(agentLoop *agent.AgentLoop) func(prompt, channel, ch
if err != nil {
return tools.ErrorResult(fmt.Sprintf("Heartbeat error: %v", err))
}
if response == "HEARTBEAT_OK" {
return tools.SilentResult("Heartbeat OK")
}
return tools.SilentResult(response)
}
}

3
web/frontend/src/components/channels/channel-forms/generic-form.tsx
View file

@ -31,9 +31,6 @@ const SECRET_FIELDS = new Set([
"password",
"nickserv_password",
"sasl_password",
"tls_ca",
"tls_cert",
"tls_key",
])
// Fields to skip in the generic form (handled by enabled toggle or internal).